쿠리이야기

주의사항airflow info 결과로 현재 사용하는 경로 (AIRFLOW_HOME, sql_alchemy_conn, logs_folder) 확인which airflow 로 실행 파일 경로 확인 → .service의 ExecStart와 동일하게 맞출 것/opt/airflow 외에 .db, logs, dags가 다른 경로에 흩어져 있는지 확인executor 확인. LocalExecutor 가 운영에 적합. airflow info | grep executor (서버 1대라 dag 및 동시실행 개수 제한 필수)sequentialExecutor -> SQLite 만 사용 가능 (?)환경변수 경로 정리 Step 0: 운영 중인 Dag 값 변경 - 운영 중단됐을 때 대비airflow.cfg[core]paralleli..

gpadmin 사용자로 airflow.service 실행 -> airflow standalone airflow info | grep AIRFLOW_HOME-> /opt/airflow /opt/airflow/airflow.cfg-> root 계정이 만든 airflow.cfg-> [core] dags_folder = /opt/airflow/dags airflow info 는 현재 로그인한 사용자의 AIRFLOW_HOME 기준 airflow.cfg 을 따른다고함.-> /home/gpadmin/airflow/airflow.cfg dagParseProcessor 는 /root/airflow/logs/dag_processor_manager 에 접근 systemd 서비스가 생성, scheduler 가 포크한 하..

section[core]전체 Airflow 시스템의 핵심 설정들 (home, dags_folder, executor, etc.)[webserver]Airflow 웹 UI와 관련된 설정 (포트, 인증 방식, expose_config 등)[scheduler]DAG 스케줄링 주기, DAG 파서 프로세서 수, heartbeat 주기 등[logging]로그 저장 경로, 로그 포맷, log level 등 설정[metrics]Prometheus 등 외부 모니터링 시스템 연동 설정[email]이메일 알림을 위한 SMTP 설정[smtp]이메일 송신 서버 설정 (이메일 기능 쓸 때 연동됨)[database]metadata DB 연결, pool, engine 설정[celery]executor가 Celery일 때 관련 설정 ..

1. Dag 파일 수정(.py)2. Scheduler 가 감지 (mtime 기준)3. queue 에 등록 (DagFileProcessorManager)4. subprocess 가 파싱 시도 (dagbag.process_file)5. DAG 객체 생성 (ex) dag=DAG(...))6. DAG 객체를 DB 에 저장하기 위해 serializedDagModel.write_dag(dag) 호출7. DAG 내부 구조를 JSON 으로 serialized8. serialized_dag 테이블에 insert/update9. web ui 에서 테이블 읽어와서 업데이트 Dag 파싱Airflow 가 .py 파일 읽고 실행 후 DAG 객체를 메모리에 올림-> dagbag.process_File() 가 처리 (메타 정보로 정..

Scheduler TimeoutThe Airflow scheduler may have enforced a task timeout (e.g., due to exceeding the execution_timeout parameter) and sent SIGTERM to stop the task.Manual TerminationA user or system admin may have manually terminated the task using airflow tasks kill or similar commands.Orphaned ProcessThe PID mismatch suggests that the task may have been restarted or orphaned, causing Airflow to..

Provider 생성은 특정 외부 시스템이나 서비스와의 통합을 위한 기능을 제공하는 패키지를 만드는 과정이다. Provider는 Airflow가 외부 시스템과 연결할 수 있도록 Hook, Operator, Sensor 등의 구성 요소를 제공한다. provider 를 만들기 위해 필요한 파일은 다음과 같다.__init__.py hook.pyoperator.pysetup.py pyproject.toml 프로젝트 생성 poetry 를 사용해 프로젝트를 생성한다.poetry new airflow-sw-provider*참고 : toml 파일을 직접 생성해서 사용하니 인식을 못하는 것 같아서 poetry 로 다시 생성했음*참고 : provider 이름을 custom_sw_provider 로 생성하고 패키지 설치한 ..

패키지를 만들기 위해 필요한 파일은 다음과 같다. __init__.py  #내용 없이 파일만 생성해주었다.Python 패키지를 만들 때 특정 디렉토리를 패키지로 인식시키기 위해 필요한 파일이다. 파일 존재 여부에 따라 해당 디렉토리가 python 패키지로 인식되며, 이로 인해 모듈 및 서브 패키지를 임포트할 수 있게 된다.*Python 3.3 이후로는 __ini__.py 이 없어도 디렉토리가 패키지로 인식되지만, 여전히 명시적으로 패키지임을 나타내기 위해 사용하는 것이 일반적이라고 한다..hook.pyoperator.pysetup.py  해당 파일들은 다음과 같은 트리 구조로 만들어준다.참고로 src 경로 안에 있는 폴더 이름은 패키징할 이름과 같아야한다./opt/airflow/dags/include/c..

airflow 에서 Custom Hook 을 생성하려면 BaseHook 을 사용한다. Hook : 외부 시스템과 상호작용하는 커넥터 생성, 공통적으로 사용하는 연결 로직이나 작업을 재사용할 수 있도록 캡슐화한다.Operator : Dag 내에서 실행되는 실제 작업을 정의한다. 다양한 Hook 을 사용해 외부 시스템과 상호작용하거나 데이터를 처리한다. Custom Hookfrom airflow.hooks.base import BaseHookfrom airflow.exceptions import AirflowExceptionimport psycopg2# hook -> 외부 시스템과 상호작용하는 커넥터 만들기# Basehook 상속 / 모든 Hook 클래스가 상속받는 기본 클래스, Airflow 의 연결 설정..

MinIO오픈 소스 오브젝트 스토리지(object storage) 솔루션으로, Amazon S3와 호환되는 인터페이스를 제공하는 고성능 분산 스토리지 시스템이다. MinIO는 대용량 데이터를 저장하고 관리하는 데 최적화되어 있으며, 클라우드 네이티브 애플리케이션과 빅데이터, AI/ML 워크로드에 적합한 스토리지 솔루션이다. 주로 Kubernetes와 같은 컨테이너 오케스트레이션 플랫폼에서 쉽게 사용할 수 있도록 설계되었다. MinIO의 구성 요소:Bucket: 데이터가 저장되는 기본 단위로, AWS S3의 버킷과 동일한 개념입니다.Object: 버킷에 저장되는 개별 데이터 파일입니다.MinIO 서버: 오브젝트 스토리지를 제공하는 서버로, 여러 노드로 구성된 클러스터에서 동작할 수 있습니다 minio cl..

Apache Spark 대규모 데이터 처리 및 분석을 위한 오픈 소스 분산 컴퓨팅 프레임워크. 대용량 데이터를 빠르게 처리할 수 있도록 설계되었으며, 특히 메모리 기반 처리(In-memory processing)를 통해 Hadoop의 MapReduce보다 훨씬 빠른 성능을 제공한다.YARN (Yet Another Resource Negotiator)Apache Hadoop의 자원 관리 프레임워크로, 클러스터의 리소스를 효율적으로 관리하고 분배하는 역할을 한다. YARN은 클러스터 내에서 애플리케이션(작업)의 실행을 관리하며, 자원 할당 및 작업 스케줄링을 담당한다.Apache Spark - YARN 관계성Apache Spark 는 YARN 위에서 실행될 수 있으며, Spark 는 YARN을 자원 관리자로..