embulk

Embulk & Digdag

Embulk

일반적인 데이터 처리 워크플로우
- 수집(Ingest/Collect) - 애플리케이션 로그, 유저 속성 정보, 광고의 인상, 서드파치쿠키
- 전처리(Enrich) - 봇의 액세스 로그 제외. IP 주소로 위치 정보 추가, user-agent의 구조화, 마스터 데이터를 사용해서 로그에 유저 속성 추가
- 분류, 집계, 분석(Model) - 데이터베이스에 추가, 분석 처리 시스템으로 전송, 압축해서 스토리지에 저장(아카이브), 통계 데이터로 기록
- 활용(Utilize) - 추천 엔진 API의 참조 데이터, 실시간 거래, BI 애플리케이션을 사용한 시각화
Fluentd는 준실시간의 로그 활용을 할 때 첫 장애물인 로그 수집의 과제를 해결하기 위해 개발되었습니다. 각종 데이터의 입력과 수집을 Input 플러그인으로 지원하고, 데이터 가공을 Filter 플러그인으로 처리하며, Output 플러그인으로 여러 가지 미들웨어나 스토리지로 저장할 수 있습니다.
- 2가지 과제
  - Fluentd의 설정 파일의 거대화와 가끔 일어나는 사업적인 사양 변경에 따라 설정 파일의 라인수가 증가하여 유지보수가 힘들게 되었습니다.
  - 스트리밍 처리에 특화된 Fluentd는 정기적으로 벌크로드를 하여 데이터 처리 워크플로우를 만들기에는 적합하지 않습니다.
데이터 워크플로우를 지원하는 도구
- 스트리밍 데이터 컬렉터 - fluentd - 액세스 로그 / 앱로그 / 서버로그
- 벌크 데이터 로더 - embulk - csv파일/ S3 / MySQL / PostgreWQL 등
- 워크플로우 관리 - digdag - ETL 처리의 자동화
Embulk
- Embulk는 Fluentd와 같이 Input/Filter/Output 플러그인을 조합해서 설정 파일을 정의합니다. 병렬 분산 처리에 대응한 성능과 재시도 제어 등에 안정성이 우수한 데이터 전송 파이프라인을 만들 수 있습니다.
- 다량의 데이터를 효율적으로 읽어서 CPU 코어를 최대한 사용해서 배치 처리하는 데 특화되어 있습니다.
- 데이터베이스와 스토리지에서 데이터를 읽어서, 임의의 처리를 한 뒤에 다른 보관 장소로 보내는 데이터의 대용량처리에 특화된 ETL 처리 도구
- Fluentd와 다른 특징으로 고속성과 트랜잭션 제어, 스키마를 사용한 데이터의 검사 기능이 있습니다.
- 대용량의 데이터를 마이크로 배치처리로 Redshift, BigQuery, Elasticsearch로 저장하는 경우라면 Fluentd 보다는 병렬처리, 처리량, 저장타이밍을 자유롭게 컨트롤할 수 있는 Embulk로 저장하는 편이 확실히 안정적입니다.

Digdag

Digdag는 워크플로우의 정의를 설정 파일로 하고 있습니다. Embulk와 임의의 셸스크립트에 임의의 변수를 넣어가며, 의존 관계순으로 직렬 및 병렬 처리로 Job을 실행할 수 있습니다.
워크플로우관리 도구로서 ETL 처리의 자동화에 도움이 됩니다.
여러 단계에서의 처리의 의존관계와 순서, 병렬실행 등을 프로그램 가능한 YAML 설정 파일을 통해 제어할 수 있는 아키텍처
여러 개의 데이터소스로부터 병렬 또는 직렬로 데이터를 읽고, 날짜별로 테이블을 만들고 저장하며 지속적인 1차 집계를 한 뒤에 그 결과를 저장하는 처리를 직관적으로 설정 파일에 설정할 수 있습니다.
기본기능
- 작업을 의존관계순으로 실행
- 과거분의 일괄실행(backfill)
- 정기 실행
- 시간 등의 변수를 포함해서 실행
- 파일이 생성되면 실행
에러처리
- 실패하면 통지
- 실패한 위치에서 재시작
상태 감시
- 실행 시간이 일정 이상이면 통지
- 작업의 실행시간의 시각화
- 실행 로그의 수집과 저장
고속화
- 작업을 병렬로 실행
- 동시 실행 작업 개수의 제어
개발지원
워크플로우의 버전 관리
- GUI로 워크플로우 개발
- 정기처리를 간단하게 실행할 수 있는 라이브러리
- Docker 이미지를 사용해서 작업 실행
Digdag에는 스케줄러를 내장하고 있어서 데몬으로 동작하는 서버 모드와 커맨드라인에서 임의로 실행하는 로컬 모드의 2가지가 있습니다. 서버 모드에서는 기밀 정보와 폴리시 파일을 커맨드라인으로 등록합니다.
Digdag을 실행하는 커맨드
- digdag run stage1_load_assets.dig
병렬 실행 수의 상한을 설정하여 Digdag의 워크플로우를 실행합니다.
- digdag run example.dig --max-task-threads 4

편리한 오퍼레이터

파일이 나타날 때까지 계속하는 s3_wait>:와 gcs_wait>:라는 오퍼레이터
AWS S3에 파일이 생성될 때까지 기다렸다가 생기면 다음 태스크로 가서 Redshift에서 가져오는 Digdag의 설정. BigQuery에서 가져오는 것도 bq_load>: 오퍼레이터를 사용하여 만들 수 있습니다.
rb>: 오퍼레이터와 py>: 오퍼레이털르 사용하면 데이터를 읽는 것 외에도 데이터의 가공도 할 수 있습니다.

$ cat tasks/__init__.py
# coding:utf-8
import digdag
import pandas as pd

class Convert(object):
  def __init(self):
    pass

  def transform(self, session_time = None, query_result=' 0'):
    input = pd.read_csv(digdag.env.params['input_csv']
    user_info = pd.read_csv(digdag.env.params['user_info']))
    combined = pd.merge(input, user_info, how='left', on=['id'])
    combined.to_csv(digdag.env.params['output_csv'], index=False, encoding='utf-8')

# Python으로 데이터 가공 처리를 하는 예제
$ cat conver_csv.dig
_export : 
  input_csv: path/to/input.csv
  output_csv: path/to/output.csv
  user_info: path/to/users.csv
  td:
    database: www_access

+extract:
  td>: queries/sample_query.sql
  download_file: ${input_csv}

+transform:
  py>: Convert.transform

+load:
  # 남은 csv 파일을 전송함(Embulk 설정은 생략)
  sh>: embulk run embulk_load_csv_bigquery.yml.liquid

Reference

http://www.yes24.com/Product/Goods/64464997
https://github.com/treasure-data/embulk
https://github.com/treasure-data/digdag

PreviousEmbulk Nextembulk_code

Last updated 1 year ago

Embulk & Digdag

Embulk

일반적인 데이터 처리 워크플로우

수집(Ingest/Collect) - 애플리케이션 로그, 유저 속성 정보, 광고의 인상, 서드파치쿠키
전처리(Enrich) - 봇의 액세스 로그 제외. IP 주소로 위치 정보 추가, user-agent의 구조화, 마스터 데이터를 사용해서 로그에 유저 속성 추가
분류, 집계, 분석(Model) - 데이터베이스에 추가, 분석 처리 시스템으로 전송, 압축해서 스토리지에 저장(아카이브), 통계 데이터로 기록
활용(Utilize) - 추천 엔진 API의 참조 데이터, 실시간 거래, BI 애플리케이션을 사용한 시각화

Fluentd는 준실시간의 로그 활용을 할 때 첫 장애물인 로그 수집의 과제를 해결하기 위해 개발되었습니다. 각종 데이터의 입력과 수집을 Input 플러그인으로 지원하고, 데이터 가공을 Filter 플러그인으로 처리하며, Output 플러그인으로 여러 가지 미들웨어나 스토리지로 저장할 수 있습니다.

2가지 과제
- Fluentd의 설정 파일의 거대화와 가끔 일어나는 사업적인 사양 변경에 따라 설정 파일의 라인수가 증가하여 유지보수가 힘들게 되었습니다.
- 스트리밍 처리에 특화된 Fluentd는 정기적으로 벌크로드를 하여 데이터 처리 워크플로우를 만들기에는 적합하지 않습니다.

데이터 워크플로우를 지원하는 도구

스트리밍 데이터 컬렉터 - fluentd - 액세스 로그 / 앱로그 / 서버로그
벌크 데이터 로더 - embulk - csv파일/ S3 / MySQL / PostgreWQL 등
워크플로우 관리 - digdag - ETL 처리의 자동화

Embulk

Embulk는 Fluentd와 같이 Input/Filter/Output 플러그인을 조합해서 설정 파일을 정의합니다. 병렬 분산 처리에 대응한 성능과 재시도 제어 등에 안정성이 우수한 데이터 전송 파이프라인을 만들 수 있습니다.
다량의 데이터를 효율적으로 읽어서 CPU 코어를 최대한 사용해서 배치 처리하는 데 특화되어 있습니다.
데이터베이스와 스토리지에서 데이터를 읽어서, 임의의 처리를 한 뒤에 다른 보관 장소로 보내는 데이터의 대용량처리에 특화된 ETL 처리 도구
Fluentd와 다른 특징으로 고속성과 트랜잭션 제어, 스키마를 사용한 데이터의 검사 기능이 있습니다.
대용량의 데이터를 마이크로 배치처리로 Redshift, BigQuery, Elasticsearch로 저장하는 경우라면 Fluentd 보다는 병렬처리, 처리량, 저장타이밍을 자유롭게 컨트롤할 수 있는 Embulk로 저장하는 편이 확실히 안정적입니다.

Digdag

Digdag는 워크플로우의 정의를 설정 파일로 하고 있습니다. Embulk와 임의의 셸스크립트에 임의의 변수를 넣어가며, 의존 관계순으로 직렬 및 병렬 처리로 Job을 실행할 수 있습니다.

워크플로우관리 도구로서 ETL 처리의 자동화에 도움이 됩니다.

여러 단계에서의 처리의 의존관계와 순서, 병렬실행 등을 프로그램 가능한 YAML 설정 파일을 통해 제어할 수 있는 아키텍처

여러 개의 데이터소스로부터 병렬 또는 직렬로 데이터를 읽고, 날짜별로 테이블을 만들고 저장하며 지속적인 1차 집계를 한 뒤에 그 결과를 저장하는 처리를 직관적으로 설정 파일에 설정할 수 있습니다.

기본기능

작업을 의존관계순으로 실행
과거분의 일괄실행(backfill)
정기 실행
시간 등의 변수를 포함해서 실행
파일이 생성되면 실행

에러처리

실패하면 통지
실패한 위치에서 재시작

상태 감시

실행 시간이 일정 이상이면 통지
작업의 실행시간의 시각화
실행 로그의 수집과 저장

고속화

작업을 병렬로 실행
동시 실행 작업 개수의 제어

개발지원

워크플로우의 버전 관리

GUI로 워크플로우 개발
정기처리를 간단하게 실행할 수 있는 라이브러리
Docker 이미지를 사용해서 작업 실행

Digdag에는 스케줄러를 내장하고 있어서 데몬으로 동작하는 서버 모드와 커맨드라인에서 임의로 실행하는 로컬 모드의 2가지가 있습니다. 서버 모드에서는 기밀 정보와 폴리시 파일을 커맨드라인으로 등록합니다.

Digdag을 실행하는 커맨드

digdag run stage1_load_assets.dig

병렬 실행 수의 상한을 설정하여 Digdag의 워크플로우를 실행합니다.

digdag run example.dig --max-task-threads 4

편리한 오퍼레이터

파일이 나타날 때까지 계속하는 s3_wait>:와 gcs_wait>:라는 오퍼레이터
AWS S3에 파일이 생성될 때까지 기다렸다가 생기면 다음 태스크로 가서 Redshift에서 가져오는 Digdag의 설정. BigQuery에서 가져오는 것도 bq_load>: 오퍼레이터를 사용하여 만들 수 있습니다.
rb>: 오퍼레이터와 py>: 오퍼레이털르 사용하면 데이터를 읽는 것 외에도 데이터의 가공도 할 수 있습니다.

$ cat tasks/__init__.py
# coding:utf-8
import digdag
import pandas as pd

class Convert(object):
  def __init(self):
    pass

  def transform(self, session_time = None, query_result=' 0'):
    input = pd.read_csv(digdag.env.params['input_csv']
    user_info = pd.read_csv(digdag.env.params['user_info']))
    combined = pd.merge(input, user_info, how='left', on=['id'])
    combined.to_csv(digdag.env.params['output_csv'], index=False, encoding='utf-8')

# Python으로 데이터 가공 처리를 하는 예제
$ cat conver_csv.dig
_export : 
  input_csv: path/to/input.csv
  output_csv: path/to/output.csv
  user_info: path/to/users.csv
  td:
    database: www_access

+extract:
  td>: queries/sample_query.sql
  download_file: ${input_csv}

+transform:
  py>: Convert.transform

+load:
  # 남은 csv 파일을 전송함(Embulk 설정은 생략)
  sh>: embulk run embulk_load_csv_bigquery.yml.liquid