개발자의 스터디 노트

chatGPT를 이용한 프레임워크중 가장 많이 사용되고 있는 프레임워크가 Langchain 일 것입니다. Langchain에서는 다양한 기능을 미리 만들어서 제공하고 있고, 다양한 검색엔진을 연계하는 Agent 또한 준비되어져 있습니다. 하지만 업무시스템에 구축되어져있는 상용검색엔진 혹은 업무시스템에 맞게 커스텀되어진 엘라스틱서치를 연계하는 맞춤형 Agent는 직접 개발해야 합니다. langchain 공식 문서의 예제 코드인 CustomSearchTool 을 활용하여 네이버 API를 호출하는 NaverSearchAPIWrapper 을 만들고 langchain과 연계 할수 있도록 해보겠습니다. NaverSearchAPIWrapper 코드를 각자 사용할 검색엔진을 호출하는 방식으로 커스터마이징 하면 손쉽게 ..

1. FileSensor 오퍼레이터 사용하기. - 특정 파일 또는 파일 세트의 존재를 확인하는 데 사용되는 센서 오퍼레이터 입니다. 이것은 일반적으로 워크 플로우가 파일이나 데이터가 있는 위치에 도달할 때까지 기다리는 시나리오에 유용합니다. from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.sensors import FileSensor from airflow.utils.dates import days_ago default_args = { 'owner': 'airflow', } with DAG( 'file_sensor_example_with_tasks', default_args=defau..

1. 의존성 정의 - with 구문을 사용하여 좀더 깔끔한 DAG를 작성할 수 있습니다. import airflow from airflow import DAG from airflow.operators.dummy import DummyOperator with DAG( dag_id="01_start", start_date=airflow.utils.dates.days_ago(3), schedule_interval="@daily", ) as dag: start = DummyOperator(task_id="start") fetch_sales = DummyOperator(task_id="fetch_sales") clean_sales = DummyOperator(task_id="clean_sales") fetch_w..

1. 태스크 콘텍스트와 Jinja 템플릿 작업 - wikipageviews에서 BashOperator를 이용하여 위키 데이터를 다운로드 하는 오퍼레이터 입니다. get_data = BashOperator( task_id="get_data", bash_command=( "curl -o /tmp/wikipageviews.gz " "https://dumps.wikimedia.org/other/pageviews/" "{{ execution_date.year }}/" "{{ execution_date.year }}-{{ '{:02}'.format(execution_date.month) }}/" "pageviews-{{ execution_date.year }}" "{{ '{:02}'.format(execution_..

Apache Airflow 기반의 데이터 파이프라인 도서의 챕터3 예제를 에러 없이 동작시키기 위하여 로컬에 fastapi를 확용해서 간단한 api를 작성 하였습니다. https://github.com/pws0601/airflow_study/tree/main/restapi/fastapi/events GitHub - pws0601/airflow_study: airflow 스터디 저장소 입니다. airflow 스터디 저장소 입니다. Contribute to pws0601/airflow_study development by creating an account on GitHub. github.com 깃 허브의 코드를 올려두었습니다. 도커 컨테이너로 배포해도 좋고 로컬에 설치해도 좋으나 docker-compose..

전체 코드 import json import pathlib import airflow.utils.dates import requests import requests.exceptions as requests_exceptions from airflow import DAG from airflow.operators.bash import BashOperator from airflow.operators.python import PythonOperator ## 실행할 DAG 정보를 정의한다. dag = DAG( dag_id="download_rocket_launches", description="Download rocket pictures of recently launched rockets.", start_date=..

이번 포스팅에서는 Apach Airflow가 무엇인지 간단히 살펴 보겠습니다. 1. 데이터 파이프 라인 - 일반적인 데이터 파이프 라인은 원하는 결과를 얻기 위하여 여러 태스크 또는 동작으로 구성됩니다. 1.1 데이터 파이프라인 그래프 - 위와 같은 그래프처럼 방향성을 가진 그래프를 방향성 그래프라고 합니다. - 이런 형태의 그래프를 방향성 비순환 그래프(Directed Acyclic Graph, DAG) 라고 부릅니다. 위 그림과 같이 태스크간의 의존성을 가지고 있을 경우 그래프를 실행할 때 순환 의존성이 문제를 발생시킬 수 있습니다. 두번째 태스크와 세번째 태스크간의 의존성으로인해 교착상태로 이어지며, 두번째, 세번째 모두 실행 될수 없기때문에 그래프가 동작하지 않습니다. 2. Airflow란. - ..

먼저 포스팅을 따라 했다면 airflow 웹 관리 화면에 샘플 DAG가 많이 등록되어 있는것을 확인 할 수 있습니다. 이 샘플 DAG가 보고 싶지 않다면 간단하게 설정을 바꿔주면 됩니다. docker-compose.yaml 파일에서 대략 59번 라인의 environment: &airflow-common-env AIRFLOW__CORE__EXECUTOR: CeleryExecutor AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow # For backward compatibility, with Airflow