저번에 elastic search에 데이터를 저장했다.
텍스트 데이터를 저장하는데 유용하다고 들었기 때문이다.
근데, elastic search가 필요 없어졌다.
1. 나는 실시간으로 데이터를 처리할 생각이 없고 하루에 한번만 데이터를 분석해 저장할 것이다.
2. 검색엔진이 필요없다.
3. elastic search는 비슷한 단어가 들어있다면 검색해서 가져온다.
예를 들어, 교육위원장 김영호를 검색하면 통일위원장 김영호도 같이 가져왔다.
그래서 다른 데이터베이스를 사용해야 하는데 nosql인 MongoDB와 rdbms인 postgres 중 고민을 하는데,
nosql인 MongoDB가 나을 거 같다는 생각이 들었다.
1. 텍스트를 elastic search와 비슷한 형식으로 저장하고 싶음
2. join이나 관계성 필요 없이 발언자와 발언 내용만을 저장하고 싶음
3. 텍스트 양이 많야 nosql의 속도가 더 빠를 거 같음
4. 특정 단어를 확인하는 NLP분석을 진행할 것임
5. 나중에 필요하다면 elastic search와 연결할 수 있음
그리고 최종으로 데이터 분석을 맞춘 데이터들만 postgres에 저장하면 될 거 같다.
MongoDB에 연결하기
먼저 MongoDB를 설치해야 한다.
https://www.mongodb.com/ko-kr/docs/manual/tutorial/install-mongodb-on-os-x/
macOS에 MongoDB Community Edition 설치 - MongoDB 매뉴얼 v8.0
설치 후 macOS가 mongod 실행을 차단할 수 있습니다. mongod을 시작할 때 개발자를 식별하거나 확인할 수 없다는 보안 오류를 수신한 경우 다음을 실행하여 mongod에 액세스 권한을 부여합니다. 시스템
www.mongodb.com
아래와 같이 하면 맥에서는 몽고 db가 실행된다.
macOS 서비스를 통해서 기본적인 몽고db를 실행할 수 있다.
백그라운드 프로세서로서 수동으로 진행하려면 다른 기능을 추가해야한다.
brew tap mongodb/brew
brew update
brew install mongodb-community@8.0
brew services start mongodb-community@8.0
mongsh
로컬 ip에서만 진행하려 한다.
어차피 로컬에서 분석을 돌리고 저장을 sql에다가 하면 되기 때문에 지금은 필요 없다고 판단 했다.
그리고 기존에 로컬에 있던 json,csv 파일은 전부 mongodb에 저장하기로 했다.
그리고 Connecting to로 Vscode에 MongoDB Extension을 추가해 연결했다.
그래서 이제부터는 지금까지 local에 파일로 저장되거나 elastic에 저장되도록 되어있는 코드들을 전부
MongoDB에 저장되도록 변경해야 한다.
db_utils라는 파일에 함수를 만들었다.
from pymongo import MongoClient
from dotenv import load_dotenv
import os
def connect_to_mongo(db_name, collection_name):
"""MongoDB 연결"""
load_dotenv()
pwd = os.getenv("MONGODB_PWD") # 나중에 암호를 추가해야할 수 있음
client = MongoClient(f"mongodb://localhost:27017/")
db = client[db_name]
collection = db[collection_name]
return client, db, collection
def save_to_mongo(data, db_name, collection_name, mongo_uri="mongodb://localhost:27017/"):
"""
데이터를 MongoDB에 저장하는 함수.
:param data: 저장할 데이터 (리스트 또는 딕셔너리)
:param db_name: 사용할 데이터베이스 이름
:param collection_name: 사용할 컬렉션 이름
:param mongo_uri: MongoDB 연결 URI (기본값: 로컬 MongoDB)
"""
try:
client, collection = connect_to_mongo(db_name,collection_name,mongo_uri)
# 데이터 저장
if isinstance(data, list): # 리스트 형태면 여러 개 저장
collection.insert_many(data)
else: # 단일 딕셔너리면 하나만 저장
collection.insert_one(data)
print(f"✅ 데이터 저장 완료! (DB: {db_name}, Collection: {collection_name})")
except Exception as e:
print(f"❌ 데이터 저장 실패: {e}")
finally:
client.close()
그 이후 데이터를 mongodb에 추가했다.
from common.api_utils import load_api_key, fetch_data
from common.db_utils import save_to_mongo
if __name__ == "__main__":
url = "https://open.assembly.go.kr/portal/openapi/nekcaiymatialqlxr"
key = load_api_key()
if key is None:
raise ValueError("❌ OPEN_GOVERMETN_API_KEY 환경 변수가 설정되지 않았습니다.")
all_data = fetch_data(url, key)
save_to_mongo(all_data, "government","congress_scedhule")
아래와 같이 얻은 Mongodb 데이터를 확인할 수 있다.
result = collection.find({},{"MEETTING_DATE":1,"_id":0}).to_list()
sql 문으로 치면 아래와 같다.
SELECT MEETTING_DATE
FROM CONGRESS_SCEDHULE;
앞에 {}에는 where절이고 뒤에 {}에는 select절이고 0과 1로 들어감 여부를 확인할 수 있다.
from pymongo import MongoClient
# MongoDB 서버에 연결
client = MongoClient("mongodb://localhost:27017/")
db = client["government"]
collection = db["congress_scedhule"]
# 데이터베이스 목록 출력
db_list = client.list_database_names()
print("✅ 현재 MongoDB 데이터베이스 목록:", db_list)
# 데이터 조회
result = collection.find({},{"MEETTING_DATE":1,"_id":0}).to_list()
print("✅ MongoDB 데이터 확인:", result)
# 연결 종료
client.close()
RDS와 연결하기
brew install postgresql
psql postgres1. postgresql을 설치하고 연결한다.
2. 그 다음에 CREATE DATABASE {원하는 데이터베이스 이름}을 추가해 db에 추가한다.
3. vscode에 sqltools라는 extension에 값을 추가한다.
4. database 아키텍쳐를 설계한다.
5. 테이블을 만들어준다.
6. psycopg2-binary를 통해 MongoDB에서 값을 가져와 테이블에 insert 한다.
CREATE TABLE CONGRESS_SCHEDULE(
meeting_date DATE PRIMARY KEY,
get_pdf BOOLEAN NOT NULL DEFAULT FALSE
);
CREATE TABLE COMITTEE_SCHEDULE(
meeting_date DATE PRIMARY KEY,
get_pdf BOOLEAN NOT NULL DEFAULT FALSE
);def get_schedule(db_name, collection_name):
"""위원회 및, 국회의원회의 날짜를 가져오는 함수"""
try:
client, db, collection = connect_to_mongo(db_name,collection_name)
result = collection.find({},{"MEETTING_DATE":1,"_id":0}).to_list()
print("✅ MongoDB 날짜 데이터 가져옴:")
return result
except Exception as e:
print(f"❌ mongodb 날짜 데이터 저장 실패: {e}")
finally:
client.close()
def connect_to_postgresql():
"""postgresql에 연결하는 함수"""
load_dotenv()
conn = psycopg2.connect(
host = os.getenv("SQL_HOST"),
dbname = os.getenv("SQL_DBNAME"),
user = os.getenv("SQL_USER"),
password = os.getenv("SQL_PWD"),
port = os.getenv("SQL_PORT")
)
return conn
def schedule_to_postgresql(schedule_list,collection_name):
try:
conn = connect_to_postgresql()
cur = conn.cursor()
for meetting_date in schedule_list:
date = meetting_date["MEETTING_DATE"]
cur.execute(f"""
INSERT INTO {collection_name} (meeting_date)
VALUES (%s)
ON CONFLICT (meeting_date) DO NOTHING
""", (date,))
conn.commit()
print("✅ postgres 데이터 저장 성공")
except Exception as e:
conn.rollback()
print(f"❌ postgres 데이터 저장 실패 : {e}")
finally:
conn.close()mongo db에서 값을 가져와 데이터를 읽어내서 sql에 저장하면 된다.
'Personal Project > 국회의원 분석 프로젝트' 카테고리의 다른 글
| python 프로젝트에서 상위 디렉토리 모듈 절대 참조 (0) | 2025.02.10 |
|---|---|
| 국회의원 회의록 정규식으로 구분하기, docker를 이용해 elastic search로 설치, 저장하기, 오류 해결하기 (0) | 2025.02.02 |
| 국회의원 회의록 데이터 가져오는 법 with Python (0) | 2025.01.31 |
| 국회의원 공약 가져오기 api 에러 - "resultMsg": "데이터 정보가 없습니다. 입력 파라미터값을 확인해주시기 바랍니다." (0) | 2025.01.25 |
| 공공 api 읽어온 후 json 파일 csv로 전환한 후 저장하기 - python (0) | 2025.01.22 |
