시나리오 1. mysql to csv 2. csv 파일 s3에 업로드 3. 실패 시 슬랙 알람 연동 ✔ 아래 링크 이어서 진행 https://heywantodo.tistory.com/20 Airflow Pipeline 생성 (1) 시나리오 1. mysql to csv 2. csv 파일 s3에 업로드 3. 실패 시 슬랙 알람 연동 1. MySQL to CSV airflow에서 mysql to csv를 실행하는 방법에는 여러가지가 있겠지만 여기선 두가지를 사용 Airflow Connection 이용 Pymys heywantodo.tistory.com 2. CSV file Upload to S3 선행작업 S3 Bucket 생성 Amazon S3 버킷 생성, 구성 및 작업 - Amazon Simple Stora..
DAG Runs DAG Run Task 인스턴스들을 DAG에 정의 된 특정 시작일에 실행하는 DAG의 인스턴스 동시에 최대 몇개의 DAG run을 돌게 허용할 것인지 셋팅하는 파라미터가 있음 Airflow Dag Task 옵션 값 Airflow Dag Task 옵션 값 Airflow에는 dag 성능을 개선할 수 있는 다양한 옵션들에는 다음과 같은 것들이 있음 parallelism 에어플로우 클러스터 전체에서 동시에 수행될 수 있는 task 인스턴스의 개수를 결 heywantodo.tistory.com 구분 내용 Status DAG 실행 결과 Dag id DAG의 이름 (식별자) Logical Date DAG이 실행 된 자체적인 논리적 시간 (기존의 execution_date) Run id DAG을 실행한..
Airflow란? Airflow Workflow를 정의하고 실행 가능한 플랫폼 반복된 작업을 자동화하기 위해 사용 각 작업들은 DAG를 통해 구조화 Airflow의 장점 연결 된 화살표 방향 순서대로 작업을 실행하고, 분기 실행과 병렬 실행이 가능함 각 작업들을 독립 된 단위로 나눠서 개발, 실행 순서를 조정하는 방법으로 모듈화 가능 데이터의 추출, 가공, 적재 기능을 나눠서 개발하고 Airflow는 이 작업들을 연결하여 파이프라인으로 배치 Airflow 구조 Airflow는 Scheduler가 DAG Directory의 작업을 가져와서 Worker에서 실행하는 형태 Scheduler Airflow의 DAG와 작업들을 모니터링하고 실행 순서와 상태 관리 Worker Airflow의 작업을 실행하는 공간 ..
테이블에서 데이터 검색하기 : SELECT 테이블에 있는 데이터를 조회하는 명령어 ex) 책 정보를 저장하는 book 테이블에 저장된 모든 데이터 검색 SELECT * FROM book; ✔ 만약 중복을 제거하고 조회를 하고 싶다면? 중복제거 : DISTINCT 뒤에 나오는 컬럼의 중복을 제거하고 보여줌 뒤에 2개 이상의 컬럼을 적으면, 한 쪽 컬럼에 중복이 있어도 다른 쪽 컬럼의 값이 다르면 다르게 취급 SELECT DISTINCT title, author FROM book; ✔ 검색 조건을 추가하고 싶다면? 조건을 추가해서 검색 : WHERE 검색하고자 하는 데이터의 조건을 설정할 수 있는 명령 SELECT * FROM book WHERE title = '돈키호테'; 🔎 여러 연산자를 사용해서 조건을..
시나리오 1. mysql to csv 2. csv 파일 s3에 업로드 3. 실패 시 슬랙 알람 연동 1. MySQL to CSV airflow에서 mysql to csv를 실행하는 방법에는 여러가지가 있겠지만 여기선 두가지를 사용 Airflow Connection 이용 Pymysql과 Pandas 이용 1-1. Airflow Connection를 이용하여 MySQL Airflow 연결 Airflow webserver에서 [Admin] - [Connections] 메뉴로 이동, +를 눌러 새연결 생성 Connection id : 연결 이름 Connection Type : MySQL (만약 없다면, 서버에서 pip install -U 'apache-airflow[mysql]' 진행) Host : 확인한 IP..
Airflow Dag Task 옵션 값 Airflow에는 dag 성능을 개선할 수 있는 다양한 옵션들에는 다음과 같은 것들이 있음 parallelism 에어플로우 클러스터 전체에서 동시에 수행될 수 있는 task 인스턴스의 개수를 결정 concurrency Dag는 concurrency 값을 초과하는 task 인스턴스를 동시에 수행할 수 없음 Dag 생성 시 concurrency를 설정할 수 있고, 따로 설정하지 않으면 dag_concurrency를 기본값으로 사용 task_concurrency 여러개의 dag_runs가 동시에 수행되는 경우 각 task 별 task 인스턴스 수를 제한 Dag안에 task가 최대 병렬적으로 몇 개 존재할 수 있는지에 대한 설정 값 max_active_runs task들의..
데이터베이스? : 여러 사람이 공유해 사용할 목적으로 통합하여 관리되는 데이터의 모음 SQL? : 데이터 베이스에 접근하고 조작하기 위한 표준 언어 데이터 베이스의 종류 관계형 데이터베이스 하나 이상의 테이블로 이루어지며, 서로 연결된 데이터를 가지고 있음 관계형 데이터베이스는 SQL을 통해 제어가 가능함 비관계형 데이터베이스 기존 관계형 데이터베이스의 한계를 뛰어넘기 위해 만들어진 새로운 형태의 데이터베이스 분산 환경에서의 데이터 처리를 더욱 빠르게 하기 위해 개발됨 데이터 베이스 스키마 스키마(Schema) 스키마 구조는 데이터베이스를 바라보는 관점으로 각각을 분리한 것 분리함으로써 얻는 이점은 독립성과 중복의 배제 스키마 종류 외부 스키마 : 각 사용자 관점 개념 스키마 : 전체 사용자 관점 내부 ..
Airflow 데이터베이스 1. 에어플로우의 DB Airflow를 설치하면 Sqlite를 Default DB로 사용하게 되기 때문에 필수 라이브러리 Sqlite는 Sequentail Executor로 순차적으로 진행하는 것만 할 수 있음 작업이 느리고 원하는 시간에 작업을 처리하지 못할 수도 있기 때문에 상용 버전에선 거의 사용하지 않음 다른 Executor를 사용하면 병렬처리가 가능해져 작업이 빨라짐, 다양한 기능 또한 있음 다른 Executor를 사용하기 위해선 DB 변경이 필요함 주로 사용하는 것이 PostgreSQL 이나 MySQL 2. Airflow DB 변경 (PostgreSQL) 공식 문서 참조 https://airflow.apache.org/docs/apache-airflow/stable/..
Airflow 실행 🔎 설치환경 AWS Amazonlinux 2 1. 자동으로 초기화하고 실행 airflow standalone 2. 수동으로 초기화하고 실행 DB 초기화 airflow db init Airflow 사용자 생성 airflow users create \ --username admin \ --firstname Jeff \ --lastname bejos \ --role Admin \ --email ceo@amazon.com \ --password PASSWD 웹 서버 & 스케줄러 시작 airflow webserver --port 8080 airflow scheduler
Airflow 설치 🔎 설치환경 AWS Amazonlinux 2 2.5.1 버전 Airflow 설치 공식 Documents 참조 https://airflow.apache.org/docs/apache-airflow/stable/installation/installing-from-pypi.html Installation from PyPI — Airflow Documentation airflow.apache.org 1. 가상환경 구성 (miniconda) 공식 사이트에서 자신의 OS 버전에 맞는 파일을 wget으로 받아옴 https://docs.conda.io/en/latest/miniconda.html Miniconda — conda documentation Miniconda is a free minimal..
