본문 바로가기
WIKI 기술 지식 베이스

Airbyte: 오픈소스 데이터 이동 인프라 (Airbyte: Open-Source Data Movement Infrastructure)

원문 보기 위키 갱신

Airbyte는 extract and load(EL) 데이터 파이프라인 구축을 위한 오픈소스 데이터 이동 인프라예요. 다양성, 확장성, 사용 편의성을 위해 설계됐어요. Airbyte의 커넥터 카탈로그는 350개 이상의 사전 구축 커넥터와 함께 "즉시 사용 가능"해요. 이 커넥터들을 사용해 몇 분 안에 소스에서 목적지로 데이터 복제를 시작할 수 있어요.

출처: Milvus 문서

본문

Airbyte의 주요 구성 요소 (Major Components of Airbyte)

1. 커넥터 카탈로그 (Connector Catalog)

  • 350+ 사전 구축 커넥터: Airbyte의 커넥터 카탈로그는 350개 이상의 사전 구축 커넥터와 함께 "즉시 사용 가능"해요. 이 커넥터들을 사용해 몇 분 안에 소스에서 목적지로 데이터 복제를 시작할 수 있어요.
  • No-Code 커넥터 빌더: No-Code Connector Builder 같은 도구를 통해 Airbyte의 기능을 확장해 사용자 지정 사용 사례를 지원할 수 있어요.

2. 플랫폼 (The Platform)

Airbyte의 플랫폼은 데이터 이동 작업을 구성하고 확장하는 데 필요한 모든 가로 서비스를 제공하며, cloud-managed 또는 self-managed로 제공돼요.

3. 사용자 인터페이스 (The User Interface)

Airbyte는 PyAirbyte(Python 라이브러리), API, Terraform Provider와 함께 API를 제공해 선호하는 도구와 인프라 관리 방식으로 통합할 수 있게 해줘요.

Airbyte의 기능으로 사용자는 유사도 검색을 위해 데이터 소스를 Milvus 클러스터에 통합할 수 있어요.

시작 전 (Before You Begin)

다음이 필요해요.

  • Zendesk 계정 (또는 데이터를 동기화하려는 다른 데이터 소스).
  • Airbyte 계정 또는 로컬 인스턴스.
  • OpenAI API 키.
  • Milvus 클러스터.
  • 로컬에 설치된 Python 3.10.

Milvus 클러스터 설정 (Set Up Milvus Cluster)

프로덕션용 K8s 클러스터를 이미 배포했다면 이 단계를 건너뛰고 Milvus Operator 배포로 직접 진행할 수 있어요. 아니라면 the steps를 따라 Milvus Operator로 Milvus 클러스터를 배포할 수 있어요.

개별 엔티티(이 경우 지원 티켓과 지식 베이스 기사)는 "collection"에 저장돼요. 클러스터가 설정된 후 컬렉션을 만들어야 해요. 적절한 이름을 선택하고 OpenAI 임베딩 서비스가 생성하는 벡터 차원과 일치하도록 Dimension을 1536으로 설정해요.

생성 후 엔드포인트와 authentication 정보를 기록해요.

Airbyte에서 연결 설정 (Set Up Connection in Airbyte)

데이터베이스가 준비됐어요. 이제 데이터를 옮겨 보아요! 이를 위해 Airbyte에서 연결을 구성해야 해요. cloud.airbyte.com에서 Airbyte 클라우드 계정에 가입하거나 the documentation에 설명된 대로 로컬 인스턴스를 시작해요.

소스 설정 (Set Up Source)

인스턴스가 실행되면 연결을 설정해야 해요. "New connection"을 클릭하고 소스로 "Zendesk Support" 커넥터를 선택해요. "Test and Save" 버튼을 클릭하면 Airbyte가 연결이 설정될 수 있는지 확인해요.

Airbyte 클라우드에서는 Authenticate 버튼을 클릭해 쉽게 인증할 수 있어요. 로컬 Airbyte 인스턴스를 사용할 때는 documentation 페이지에 설명된 지침을 따라요.

목적지 설정 (Set Up Destination)

모든 것이 올바르게 작동하면 다음 단계는 데이터를 옮길 목적지를 설정하는 것이에요. 여기서 "Milvus" 커넥터를 선택해요.

Milvus 커넥터는 세 가지 작업을 해요.

  • 청킹과 포맷팅 (Chunking and Formatting) - Zendesk 레코드를 텍스트와 메타데이터로 분할해요. 텍스트가 지정된 청크 크기보다 크면 레코드가 여러 부분으로 분리되어 개별적으로 컬렉션에 로드돼요. 텍스트 분할(또는 청킹)은 큰 지원 티켓이나 지식 기사 같은 경우에 발생할 수 있어요. 텍스트를 분할하면 검색이 항상 유용한 결과를 산출하도록 보장할 수 있어요.

청크 크기를 1000 토큰으로 정하고 텍스트 필드로 body, title, description, subject를 사용해요. 이 필드들은 Zendesk에서 받는 데이터에 존재할 것이기 때문이에요.

  • 임베딩 (Embedding) - 머신러닝 모델을 사용해 처리 부분에서 생성한 텍스트 청크를 시맨틱 유사성을 검색할 수 있는 벡터 임베딩으로 변환해요. 임베딩을 만들려면 OpenAI API 키를 제공해야 해요. Airbyte가 각 청크를 OpenAI에 보내고 결과 벡터를 Milvus 클러스터에 로드되는 엔티티에 추가해요.
  • 인덱싱 (Indexing) - 청크를 벡터화한 후 데이터베이스에 로드할 수 있어요. 이를 위해 Milvus 클러스터에서 클러스터와 컬렉션을 설정할 때 받은 정보를 삽입해요. "Test and save"를 클릭하면 모든 것이 올바르게 정렬되었는지(유효한 자격 증명, 컬렉션 존재, 컬렉션이 구성된 임베딩과 같은 벡터 차원을 가지는지 등) 확인해요.

스트림 sync 흐름 설정 (Set up stream sync flow)

데이터가 흐를 준비가 되기 전 마지막 단계는 동기화할 "streams"를 선택하는 것이에요. 스트림은 소스의 레코드 모음이에요. Zendesk는 우리 사용 사례와 관련 없는 많은 스트림을 지원하므로 대역폭을 절약하고 검색에 관련 정보만 표시되도록 "tickets"와 "articles"만 선택하고 다른 모든 것은 비활성화해요.

스트림 이름을 클릭하면 소스에서 추출할 필드를 선택할 수 있어요. "Incremental | Append + Deduped" sync 모드는 최소한의 데이터(마지막 실행 이후 변경된 기사와 티켓만)를 전송하면서 이후 연결 실행이 Zendesk와 Milvus를 동기화 상태로 유지한다는 뜻이에요.

연결이 설정되는 즉시 Airbyte가 데이터 동기화를 시작해요. Milvus 컬렉션에 나타나는 데 몇 분이 걸릴 수 있어요.

복제 빈도를 선택하면 Airbyte가 정기적으로 실행되어 Zendesk 기사의 변경과 새로 생성된 이슈로 Milvus 컬렉션을 최신 상태로 유지해요.

흐름 확인 (Check flow)

Milvus 클러스터 UI에서 playground로 이동해 _ab_stream == "tickets"으로 설정된 필터로 "Query Data" 쿼리를 실행하면 컬렉션에서 데이터가 어떻게 구조화되는지 확인할 수 있어요.

Result 보기에서 볼 수 있듯이 Zendesk에서 온 각 레코드는 지정된 모든 메타데이터와 함께 Milvus에서 별도의 엔티티로 저장돼요. 임베딩의 기반이 되는 텍스트 청크는 "text" 속성으로 표시돼요. 이것이 OpenAI로 임베딩되고 우리가 검색할 텍스트예요.

컬렉션을 쿼리하는 Streamlit 앱 구축 (Build Streamlit app querying the collection)

데이터가 준비됐어요. 이제 데이터를 사용할 애플리케이션을 구축해야 해요. 이 경우 애플리케이션은 사용자가 지원 사례를 제출하는 간단한 지원 양식이에요. 사용자가 제출하면 두 가지 작업을 해요.

  • 같은 조직의 사용자가 제출한 유사한 티켓 검색.
  • 사용자와 관련 있을 수 있는 지식 기반 기사 검색.

두 경우 모두 OpenAI 임베딩을 사용한 시맨틱 검색을 활용해요. 이를 위해 사용자가 입력한 문제 설명도 임베딩되어 Milvus 클러스터에서 유사한 엔티티를 검색하는 데 사용돼요. 관련 결과가 있으면 양식 아래에 표시돼요.

UI 환경 설정 (Set up UI environment)

Streamlit으로 애플리케이션을 구현하므로 로컬 Python 설치가 필요해요.

먼저 Streamlit, Milvus 클라이언트 라이브러리, OpenAI 클라이언트 라이브러리를 로컬에 설치해요.

pip install streamlit pymilvus openai

기본 지원 양식을 렌더링하려면 basic_support_form.py라는 파이썬 파일을 만들어요.

import streamlit as st

with st.form("my_form"):
    st.write("Submit a support case")
    text_val = st.text_area("Describe your problem")

    submitted = st.form_submit_button("Submit")
    if submitted:
        # TODO check for related support cases and articles
        st.write("Submitted!")

애플리케이션을 실행하려면 Streamlit run을 사용해요.

streamlit run basic_support_form.py

이렇게 하면 기본 양식이 렌더링돼요.

이 예시의 코드는 GitHub에서도 찾을 수 있어요.

백엔드 쿼리 서비스 설정 (Set up backend query service)

다음으로 관련될 수 있는 기존 공개 티켓이 있는지 확인해요. 이를 위해 사용자가 입력한 텍스트를 OpenAI로 임베딩한 뒤 컬렉션에서 유사도 검색을 수행하고, 여전히 공개된 티켓으로 필터링해요. 제공된 티켓과 기존 티켓 사이의 거리가 매우 낮은 티켓이 있다면 사용자에게 알리고 제출하지 않아요.

import streamlit as st
import os
import pymilvus
import openai

with st.form("my_form"):
    st.write("Submit a support case")
    text_val = st.text_area("Describe your problem?")

    submitted = st.form_submit_button("Submit")
    if submitted:
        import os
        import pymilvus
        import openai

        org_id = 360033549136 # TODO Load from customer login data

        pymilvus.connections.connect(uri=os.environ["MILVUS_URL"], token=os.environ["MILVUS_TOKEN"])
        collection = pymilvus.Collection("zendesk")

        embedding = openai.Embedding.create(input=text_val, model="text-embedding-ada-002")['data'][0]['embedding']

        results = collection.search(data=[embedding], anns_field="vector", param={}, limit=2, output_fields=["_id", "subject", "description"], expr=f'status == "new" and organization_id == {org_id}')

        st.write(results[0])
        if len(results[0]) > 0 and results[0].distances[0] < 0.35:
            matching_ticket = results[0][0].entity
            st.write(f"This case seems very similar to {matching_ticket.get('subject')} (id #{matching_ticket.get('_id')}). Make sure it has not been submitted before")
        else:
            st.write("Submitted!")

여기서 여러 가지가 일어나요.

  • Milvus 클러스터에 대한 연결이 설정돼요.
  • OpenAI 서비스를 사용해 사용자가 입력한 설명의 임베딩을 생성해요.
  • 티켓 상태와 조직 ID로 결과를 필터링해 유사도 검색을 수행해요(같은 조직의 공개 티켓만 관련 있기 때문).
  • 결과가 있고 기존 티켓의 임베딩 벡터와 새로 입력된 텍스트 사이의 거리가 특정 임계값 아래면 그 사실을 알려줘요.

새 앱을 실행하려면 먼저 OpenAI와 Milvus용 환경 변수를 설정해야 해요.

export MILVUS_TOKEN=...
export MILVUS_URL=https://...
export OPENAI_API_KEY=sk-...

streamlit run app.py

이미 존재하는 티켓을 제출하려고 하면 결과가 다음과 같이 보일 거예요.

이 예시의 코드는 GitHub에서도 찾을 수 있어요.

더 관련성 높은 정보 표시 (Show more relevant information)

최종 버전에 숨겨진 초록색 디버그 출력에서 볼 수 있듯이 두 티켓이 검색과 일치했어요(state new, 현재 조직, 임베딩 벡터에 가까움). 그러나 첫 번째(관련)가 두 번째(이 상황에서 관련 없는)보다 높은 순위를 차지했으며, 이는 더 낮은 거리 값으로 반영돼요. 이 관계는 일반 전문 검색처럼 단어를 직접 일치시키지 않고 임베딩 벡터에 포착돼요.

마무리로, 티켓 제출 후 사용자에게 최대한 많은 관련 정보를 제공하기 위해 유용한 정보를 표시해요.

이를 위해 티켓이 제출된 후 두 번째 검색을 수행해 최고 일치 지식 베이스 기사를 가져와요.

   ......
   
        else:
            # TODO Actually send out the ticket
            st.write("Submitted!")
            article_results = collection.search(data=[embedding], anns_field="vector", param={}, limit=5, output_fields=["title", "html_url"], expr=f'_ab_stream == "articles"')
            st.write(article_results[0])
            if len(article_results[0]) > 0:
                st.write("We also found some articles that might help you:")
                for hit in article_results[0]:
                    if hit.distance < 0.362:
                        st.write(f"* [{hit.entity.get('title')}]({hit.entity.get('html_url')})")

높은 유사도 점수의 공개 지원 티켓이 없으면 새 티켓이 제출되고 관련 지식 기사가 아래에 표시돼요.

이 예시의 코드는 Github에서도 찾을 수 있어요.

결론 (Conclusion)

여기서 보여 주는 UI는 실제 지원 양식이 아니라 사용 사례를 설명하기 위한 예시지만, Airbyte와 Milvus의 결합은 매우 강력해요. 다양한 소스(Postgres 같은 데이터베이스, Zendesk나 GitHub 같은 API, Airbyte의 SDK나 시각적 커넥터 빌더로 구축한 완전히 사용자 지정 소스까지)에서 텍스트를 로드하고 데이터를 엄청난 양으로 확장할 수 있는 강력한 벡터 검색 엔진인 Milvus에 임베딩된 형태로 인덱싱하기 쉽게 만들어줘요.

Airbyte와 Milvus는 오픈소스이며 인프라에서 완전히 무료로 사용할 수 있고, 원한다면 운영을 오프로드할 수 있는 클라우드 제공도 있어요.

이 기사에서 설명한 고전적 시맨틱 검색 사용 사례를 넘어, 일반 설비는 RAG 방법(Retrieval Augmented Generation)을 사용한 질문-답변 챗봇, 추천 시스템 구축, 또는 광고를 더 관련성 있고 효율적으로 만드는 데도 사용할 수 있어요.

더 알아보기 (Learn more)

함께 읽는 실무 이야기

우리 조직에도 적용하고 싶다면, AI 시스템 구축 서비스를 살펴보세요.