Airflow は、バッチ データ処理プロセスを便利かつ迅速に開発および維持するためのツールです。

こんにちは、ハブル！ この記事では、企業の DWH や DataLake のインフラストラクチャなどでバッチ データ処理プロセスを開発するための XNUMX つの優れたツールについて説明したいと思います。 Apache Airflow（以下、Airflow）についてお話します。 Habré については不当に注目が集まっていないため、本文では、ETL/ELT プロセスのスケジューラを選択する際に、少なくとも Airflow に注目する価値があることを説得していきたいと思います。

以前、私は Tinkoff Bank で働いていたときに、DWH をテーマにした一連の記事を書きました。 現在、私は Mail.Ru グループ チームの一員となり、ゲーム分野のデータ分析用のプラットフォームを開発しています。 実際、ニュースや興味深いソリューションが登場するたびに、私のチームと私はここでデータ分析用のプラットフォームについて話すつもりです。

プロローグ

それでは、始めましょう。 エアフローとは何ですか? ここは図書館です（または ライブラリのセット) 作業プロセスを開発、計画、監視するため。 Airflow の主な特徴: Python コードを使用してプロセスを記述 (開発) します。 これには、プロジェクトと開発を整理する上で多くの利点があります。本質的に、(たとえば) ETL プロジェクトは単なる Python プロジェクトであり、インフラストラクチャの詳細、チームの規模、およびその他の要件。 楽器的にはすべてがシンプルです。 たとえば、PyCharm + Git を使用します。 素晴らしくてとても便利ですよ！

次に、Airflow の主要なエンティティを見てみましょう。 それらの本質と目的を理解することで、プロセス アーキテクチャを最適に編成できます。 おそらく主体となるのは有向非巡回グラフ（以下、DAG）でしょう。

DAG

DAG は、特定のスケジュールに従って厳密に定義された順序で完了する必要があるタスクの意味のある関連付けです。 Airflow は、DAG やその他のエンティティを操作するための便利な Web インターフェイスを提供します。

DAG は次のようになります。

開発者は、DAG を設計するときに、DAG 内のタスクを構築する一連のオペレーターを規定します。 ここで、もう XNUMX つの重要なエンティティである Airflow Operator について説明します。

演算子

オペレーターは、ジョブ インスタンスの作成に基づいたエンティティであり、ジョブ インスタンスの実行中に何が起こるかを記述します。 GitHub からの Airflow リリース すぐに使用できる一連の演算子がすでに含まれています。 例:

• BashOperator - bash コマンドを実行するためのオペレーター。
• PythonOperator - Python コードを呼び出すための演算子。
• EmailOperator — 電子メールを送信するための演算子。
• HTTPOperator - http リクエストを処理するための演算子。
• SqlOperator - SQL コードを実行するための演算子。
• センサーはイベント（必要な時刻の到来、必要なファイルの出現、データベースの行、APIからの応答など）を待つためのオペレーターです。

より具体的なオペレーターとしては、DockerOperator、HiveOperator、S3FileTransferOperator、PrestoToMysqlOperator、SlackOperator があります。

独自の特性に基づいてオペレーターを開発し、プロジェクトで使用することもできます。 たとえば、MongoDB から Hive にドキュメントをエクスポートするためのオペレーターである MongoDBToHiveViaHdfsTransfer と、それを操作するためのいくつかのオペレーターを作成しました。 クリックハウス: CHLoadFromHiveOperator および CHTableLoaderOperator。 基本的に、プロジェクトで基本的なステートメントに基づいて構築されたコードが頻繁に使用されるようになったら、すぐにそれを新しいステートメントに構築することを検討できます。 これにより、さらなる開発が簡素化され、プロジェクト内の演算子のライブラリが拡張されます。

次に、タスクのこれらすべてのインスタンスを実行する必要があります。次に、スケジューラについて説明します。

プラハ

Airflow のタスク スケジューラは以下に基づいて構築されています セロリ。 Celery は、キューを整理し、タスクの非同期および分散実行を可能にする Python ライブラリです。 Airflow 側では、すべてのタスクがプールに分割されます。 プールは手動で作成されます。 通常、その目的は、ソースを操作する作業負荷を制限すること、または DWH 内のタスクを典型化することです。 プールは Web インターフェイス経由で管理できます。

各プールにはスロット数に制限があります。 DAG を作成するとき、DAG にはプールが与えられます。

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

DAG レベルで定義されたプールは、タスク レベルでオーバーライドできます。
別のプロセスであるスケジューラは、Airflow 内のすべてのタスクのスケジュールを担当します。 実際、スケジューラは、実行するタスクを設定するすべての仕組みを処理します。 タスクは実行される前にいくつかの段階を経ます。

1. DAG 内の以前のタスクは完了しているため、新しいタスクをキューに入れることができます。
2. キューはタスクの優先度に応じてソートされ（優先度の制御も可能）、プールに空きスロットがあればタスクを実行できます。
3. 空いているワーカー セロリがある場合、タスクはそこに送信されます。 問題でプログラムした作業が、XNUMX つまたは別の演算子を使用して開始されます。

十分シンプルです。

スケジューラは、すべての DAG のセットと DAG 内のすべてのタスクで実行されます。

スケジューラーが DAG との連携を開始するには、DAG がスケジュールを設定する必要があります。

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

既製のプリセットのセットがあります。 @once, @hourly, @daily, @weekly, @monthly, @yearly.

cron 式を使用することもできます。

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

実行日

Airflow がどのように機能するかを理解するには、DAG の実行日を理解することが重要です。 Airflow では、DAG には実行日ディメンションがあります。つまり、DAG の作業スケジュールに応じて、タスク インスタンスが実行日ごとに作成されます。 また、実行日ごとにタスクを再実行できます。また、たとえば、DAG は複数の実行日で同時に動作することもできます。 これはここで明確に示されています。

残念ながら (あるいは幸いなことに、状況によって異なりますが)、DAG 内のタスクの実装が修正された場合、調整を考慮して前の実行日での実行が続行されます。 これは、新しいアルゴリズムを使用して過去の期間のデータを再計算する必要がある場合には適していますが、結果の再現性が失われるため好ましくありません (もちろん、必要なバージョンのソース コードを Git から返して何を計算するかをわざわざ面倒に思う人はいません)必要なのは一度だけ、必要な方法で）。

タスクの生成

DAG の実装は Python のコードであるため、たとえばシャーディングされたソースを使用する場合にコードの量を削減する非常に便利な方法があります。 ソースとして XNUMX つの MySQL シャードがあるとします。それぞれのシャードにアクセスして、データを取得する必要があります。 しかも独立かつ並行して。 DAG 内の Python コードは次のようになります。

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='/data/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:], },
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

DAG は次のようになります。

この場合、設定を調整して DAG を更新するだけで、シャードを追加または削除できます。 快適！

また、より複雑なコード生成を使用することもできます。たとえば、データベースの形式でソースを操作したり、テーブル構造やテーブルを操作するためのアルゴリズムを記述したり、DWH インフラストラクチャの機能を考慮してプロセスを生成したりすることもできます。 N 個のテーブルをストレージにロードするためのものです。 または、たとえば、リスト形式のパラメーターの操作をサポートしていない API を操作する場合、このリストから DAG に N 個のタスクを生成し、API 内のリクエストの並列処理をプールに制限し、必要なデータをAPIから取得します。 フレキシブル！

リポジトリ

Airflow には、タスク、DAG、接続設定、グローバル変数などの状態を保存する独自のバックエンド リポジトリ、データベース (MySQL または Postgres を使用できます。Postgres もあります) があります。ここで言えることは、 Airflow のリポジトリは非常にシンプル (約 20 テーブル) で、その上に独自のプロセスを構築したい場合に便利です。 Informatica リポジトリには 100500 のテーブルがあり、クエリの作成方法を理解するまでに長い時間をかけて研究する必要があったことを覚えています。

監視

リポジトリがシンプルであるため、自分にとって便利なタスク監視プロセスを構築できます。 Zeppelin でメモ帳を使用して、タスクのステータスを確認します。

これは、Airflow 自体の Web インターフェイスである可能性もあります。

Airflow コードはオープンソースであるため、Telegram にアラートを追加しました。 タスクの実行中の各インスタンスでエラーが発生すると、開発およびサポート チーム全体が構成される Telegram のグループにスパムが送信されます。

Telegram (必要な場合) を通じて迅速な応答を受け取り、Zeppelin を通じて Airflow のタスクの全体像を受け取ります。

合計で

Airflow は主にオープンソースであり、それに奇跡を期待すべきではありません。 有効なソリューションを構築するために時間と労力を費やす準備をしてください。 目標は達成可能です、信じてください、それだけの価値があります。 開発のスピード、柔軟性、新しいプロセスの追加の容易さ - きっと気に入っていただけるでしょう。 もちろん、プロジェクトの構成やエアフロー自体の安定性に十分な注意を払う必要があります。奇跡は起こりません。

現在、Airflow が毎日機能しています 約6,5千タスク。 彼らは性格が全く異なります。 多くの異なる非常に特殊なソースからメイン DWH にデータをロードするタスクがあり、メイン DWH 内のストアフロントを計算するタスクがあり、高速 DWH にデータを公開するタスクがあり、非常に多くの異なるタスクがあります - そして Airflow毎日毎日それらをすべて噛み砕きます。 数字で言うとこれです 2,3千 DWH (Hadoop) 内のさまざまな複雑さの ELT タスク、約 2,5 のデータベース 情報筋、これはからのチームです 4 人の ETL 開発者、DWH での ETL データ処理と DWH 内での ELT データ処理、そしてもちろんそれ以外にも分かれています。 XNUMX 人の管理者、サービスのインフラストラクチャを担当します。

将来の計画

プロセスの数は必然的に増加しており、Airflow インフラストラクチャに関して行う主な作業はスケーリングです。 Airflow クラスターを構築し、Celery ワーカーに XNUMX 組のレッグを割り当て、ジョブ スケジューリング プロセスとリポジトリを備えた自己複製ヘッドを作成したいと考えています。

フィナーレ

もちろん、これが Airflow について伝えたいすべてではありませんが、主要なポイントを強調してみました。 食べると食欲が湧いてきます。ぜひ試してみてください。きっと気に入っていただけるでしょう :)

出所： habr.com