[RESUMAI] 4. 데이터 크롤링과 vectorDB 저장

GITHUB

GitHub - resum-ai/resume-crawler: Resumai의 크롤러 코드를 작성한 레포지토리입니다.

 

지난 글에 이어 데이터를 모두 크롤링하고, pinecone에 저장하는 작업을 진행했다. 사실 어떤 DB를 사용할까 고민했던 이유는, vectorDB에 vector만 저장할 수 있는 줄 알았기 때문이었다. 하지만 찾아보니, 정말 다행스럽게도 metadata로 string, integer, bool 정도의 기본 데이터를 저장할 수 있다는 것을 알게 되었다. 그렇게 pinecone을 사용하여 데이터를 저장했다. 아래는 구현 과정이다.

---

1. 크롤링 후 sqlite3에 저장

지난 글에는 잡코리아에서 '지원 회사, 지원 시기, 스펙, 질문, 답변' 이렇게 5가지의 항목을 저장하려고 했었다. 그러나, 어차피 벡터 유사도를 구할 거였으면 질문과 답변 쌍만 잘 짜여져 있으면 될 것 같다고 생각하여, 나머지를 모두 삭제하고 질문, 답변 리스트만 크롤링하는 코드로 바꾸었다.

 

#jobkorea.py

def self_introduction_crawl(driver: webdriver.Chrome, file_url):
    question_list = []
    answer_list = []

    driver.get(file_url)

    paper = driver.find_element(By.CLASS_NAME, "qnaLists")
    questions = paper.find_elements(By.TAG_NAME, "dt")
    # print("회사 질문")
    for index in questions:
        question = index.find_element(By.CLASS_NAME, "tx")
        if question.text == "":
            index.find_element(By.TAG_NAME, "button").click()
            question = index.find_element(By.CLASS_NAME, "tx")
            question_list.append(question.text)
        else:
            question_list.append(question.text)
    # print(question_list)
    driver.implicitly_wait(3)

    answers = paper.find_elements(By.TAG_NAME, "dd")
    driver.implicitly_wait(3)
    # print("답변")
    for index in range(len(answers)):
        answer = answers[index].find_element(By.CLASS_NAME, "tx")
        if answer.text == "":
            questions[index].find_element(By.TAG_NAME, "button").click()
            answer = answers[index].find_element(By.CLASS_NAME, "tx")
        answer_list.append(answer.text)
    # print(answer_list)

    return {
        "question_list": question_list,
        "answer_list": answer_list
    }

 

이 코드를 이용해서 quesiton_list, answer_list를 받아와서, 임베딩 및 전처리를 한 후, vectorDB에 넣을 생각이었지만,,

이 과정을 한꺼번에 하는 것은 너무 위험한 것 같다고 생각했다.

 

준비된 자소서 링크는 약 950개였고, 각 링크마다 평균 3~4개의 질문이 있으므로, 최소 3000개의 질문-답변 쌍이 나올 것인데, 크롤링 또는 데이터 적재 도중 에러가 발생하면 끊긴 지점부터 다시 해야 하기 때문이다. 

 

이에, 안정성을 위해 sqlite3라는 경량 DB에 먼저 크롤링 데이터를 저장하고, 이후에 vectorDB에 적재하는 방향으로 결정했다. 

그래서 아래와 같이 코드를 작성했다.

# 데이터베이스 연결
conn = sqlite3.connect('crawling_data.db')
cursor = conn.cursor()
cursor.execute('''CREATE TABLE IF NOT EXISTS data
                  (Question TEXT, Answer TEXT)''')

# 데이터 저장 함수
def save_to_db(question, answer):
    # 중복 검사
    cursor.execute("SELECT * FROM data WHERE Question = ? AND Answer = ?", (question, answer))
    if cursor.fetchone() is None:
        cursor.execute("INSERT INTO data (Question, Answer) VALUES (?, ?)", (question, answer))
        conn.commit()

# 에러난 후 다시 시도하여 처음 n줄을 건너뛰기 위함.
crawled = 499

for _ in range(crawled):
    next(file)
    
read_count = crawled
qa_data = []

while True:
    file_url = file.readline()
    print(read_count, "번째 줄")

    if file_url == "":
        break

    try:
        qa_result = jobkorea.self_introduction_crawl(driver=driver, file_url=file_url)
        question_list = qa_result["question_list"]
        answer_list = qa_result["answer_list"]

        for index in range(len(question_list)):
            question = question_list[index]
            answer = answer_list[index]

            # DB에 저장
            save_to_db(question, answer)

            qa_data.append({
                "Question": question,
                "Answer": answer
            })

    except Exception as e:
        print(f"{read_count}번째에서 다음 에러가 발생했습니다: {e}")

    read_count += 1

 

이렇게 전반부에는 데이터베이스 연결, table 생성하여, 중복된 데이터가 있는지에 대해 검사하는 코드도 작성했다.

이후, Question, Answer 쌍을 save_to_db() 함수를 이용해서 데이터베이스에 저장했다.

 

다만, 이 과정에서 SSL 인증 때문에 에러가 발생하여 끊기는 경우가 있었는데, 이는 print() 문으로 계속 로깅하다가 끊긴 만큼 뛰어넘고 저장하는 방식으로 진행했다. 

 

이렇게 성공적으로 sqlite3에 데이터를 모두 저장할 수 있었다.

총 3800개..생각보다 많았다!

 

---

2. Pinecone 저장: 공식문서부터 효율적인 저장까지

이제 위에서 저장한 데이터들을 불러와 pinecone에 저장하는 작업을 거쳐야 했다. 일단 pinecone에 대해서 너무 몰라서, 공식 문서를 공부해 보았다.

우선 pinecone api key를 발급받아 init 시키고, 연결하여 upsert() 함수를 통해 데이터를 넣는 방식인 것 같았다. (물론 upsert 전에 벡터로 바꾸어야 했다.)

그리고 중요한 메타데이터. 다음 항목들만 가능하다고 하더라.

사실 쓸건 number, string 밖에 없었기에, 이 정도면 완전 충분했다.

 

마지막으로 query로 조회하면 다음처럼 나온다고 했다.

return 되는 항목을 보니, 딱 필요한 항목들만 있어서 마음에 들었다.

 

이제 이걸 코드로 구현해 보자.

먼저, 읽은 것처럼 init 하고, 저장한 데이터베이스로부터 데이터를 가져와 datas라고 이름 붙였다.

with open("secrets.json", "r") as file:
   secrets = json.load(file)

pinecone.init(api_key=secrets["PINECONE_API_KEY"], environment="gcp-starter")

index = pinecone.Index('resumai-self-introduction-index')

#데이터 로딩
conn = sqlite3.connect('crawling_data.db')
cursor = conn.cursor()

cursor.execute("SELECT * FROM data")
datas = cursor.fetchall()

conn.close()

 

그리고 print(len(data)를 해봤는데 3800개가 나왔다. 

이를 보고 생각이 든 건, '과연 이 3800개의 데이터를 for문을 통해 하나하나 upsert 해야 하나?'였다. 

일단 너무 비효율적이라고 생각하기도 했고, 요청 횟수에 제한에 있을 것 같기도 했다.

 

아니나 다를까, batch로 만들어 저장하는 방식이 공식문서에 알려져 있었다.

그것을 참고하여 코드를 다음과 같이 작성했다.

with pinecone.Index('resumai-self-introduction-index', pool_threads=30) as pinecone_index:
    async_results = [
        pinecone_index.upsert(vectors=ids_vectors_chunk, async_req=True)
        for ids_vectors_chunk in chunks(generate_data(datas), batch_size=100)
    ]
    [async_result.get() for async_result in async_results]

 

여기서 chunks와 generate_data는 lib 파일 내에 저장했다. 코드는 다음과 같다:

def chunks(iterable, batch_size=100):
    it = iter(iterable)
    chunk = tuple(itertools.islice(it, batch_size))
    while chunk:
        yield chunk
        chunk = tuple(itertools.islice(it, batch_size))

def generate_data(datas):
    for index, data in enumerate(datas):
        question = data[0]
        answer = data[1]
        qa = question + "\n\n" + answer
        qa_embedding = get_embedding(qa)
        qa_index = index + 1
        print(qa_index)
        yield (
            str(qa_index),
            qa_embedding,
            {
                "question": question,
                "answer": answer,
                "qa_index": qa_index
            }
        )

 

이렇게 작성함으로써, 100개의 묶음으로 된 배치를 38번 (3800 / 100 = 38)만 보내면 됐었고, 

이 또한 비동기 요청을 사용함으로써, 여러 업로드 요청을 동시에 보낼 수 있으며, 각 요청이 독립적으로 처리되므로 전체 업로드 시간을 단축할 수 있었다.

 

최종적으로, pinecone에 저장된 데이터는 다음과 같았다.

pinecone에 대해 잘은 몰라서 이것저것 만져봤는데, 너무 신기했다.

이렇게, 쿼리를 직접 날려보고, topK를 가져오는 것도 할 수 있었다. (약간 postman 같은 느낌 ???)

아무튼 딱 원하는 데이터베이스를 딱 알맞게 찾아서 다행이었고, 생각보다 쉽게 저장해서 다행이었다.

---

3. 정리 및 아쉬운 점

이렇게 pinecone에 데이터를 저장하는 것까지 끝내봤다.

프로젝트가 어느 정도 완성되고 나면, 최신 데이터를 가져오는 크롤러부터, pinecone에 저장하는 것까지 자동화해보고 싶다고 생각했다. 그럼 최신 데이터가 효율적으로 잘 적재될 수 있지 않을까..ㅎㅎ

 

아쉬웠던 점은, pinecone에 3800개의 데이터를 보내는 사이 할 게 없어서 sqlite에 저장했던 데이터들을 좀 둘러봤는데,

 

무슨 이상한 문자들이 있었고, 마지막에는 글자수가 몇 개인지 세어 주는 것도 함께 크롤링돼 있었다.

아... 데이터를 잘 확인했어야 했는데,,, 너무 아쉽지만 지금 당장은 넣어놓은 데이터로 진행하고, 좀 추후에 전처리하는 로직도 짜보기로 했다.

 

다음부터는 무조건 크롤링하고 데이터를 먼저 보는 습관을 들여야겠다 !!!!!