はじめに

Airflow関連の作業の備忘録

対象バージョン

１系

Airflowとは
- ワークフローをpythonで書いたDAGを作成し、定常実行するものです。
- https://airflow.apache.org
- 類似でdigdag、jenkinsなどあります。
Airflow source code
- https://github.com/apache/airflow
アーキテクチャ
- sql alchemy
  - https://qiita.com/bokotomo/items/a762b1bc0f192a55eae8
- 図
  - https://cloud.google.com/composer/docs/concepts/overview?hl=ja#architecture
    - celeryのredis利用用途
      - celery message broker (タスクキューイング用 redis, RabbitMQ等が使える)
- Executor
  - メッセージのキューイングするプロセスです。ExecutorとSchedulerは密に結合しており、スケジュールされた各タスクを実際に実行するWorkerプロセスを決定します。次のような異なるタイプのExecutorがあります。
    - SequentialExecutor: Schedulerプロセスと同じマシン上でタスクをシーケンシャルに処理する
    - LocalExecutor: Schedulerプロセスと同じマシン上でタスクを並列処理する
    - CeleryExecutor: Workerマシンのクラスタにタスクを分散して並列処理する
      - キューイングから実行まで
        
        実行部分
        
        execute_async がコールされます。キューイングも指定がなければデフォルト
        
        https://github.com/apache/airflow/blob/1.10.2/airflow/executors/celery_executor.py#L46-L86

airflow.cfg name	Environment Variable	Default Value
parallelism	AIRFLOW__CORE__PARALLELISM	32
dag_concurrency	AIRFLOW__CORE__DAG_CONCURRENCY	16
worker_concurrency	AIRFLOW__CELERY__WORKER_CONCURRENCY	16
max_threads	AIRFLOW__SCHEDULER__MAX_THREADS	2

parallelism
- タスク同時実行数。これは、全ての実行しているDAGを通しての制限値になります。
dag_concurrency
- 特定DAG内のタスク同時実行数
- たとえ、50のタスクを持ったDAGでも同時実行数はこのパラメータに制限されます。
worker_concurrency
- 一つのワーカーで実行されるタスクの数。４つワーカーがあれば、16のworker concurrencyで最大64タスクを同時に実行できます。

タスクに優先度をつけたい場合
- queue、priority_weightを利用する
- 設定できるのはCPU優先度部分のみ？メモリなどはなさそう。
- https://airflow.apache.org/concepts.html