エアフロー-バッチデータ処理プロセスを便利かつ迅速に開発および保守するためのツール

こんにちは、Habr！ この記事では、たとえば企業のDWHインフラストラクチャやDataLakeでバッチデータ処理プロセスを開発するための優れたツールについて説明します。 Apache Airflow（以降Airflow）についてです。 彼はHabréに不当に注意を奪われており、主な部分では、ETL / ELTプロセスのスケジューラーを選択する際に少なくともAirflowを検討する必要があることを納得させようとします。

私は以前、Tinkoff Bankで働いていたときにDWHのトピックに関する一連の記事を書きました。 現在、私はMail.Ru Groupチームの一員となり、ゲームの方向性に関するデータ分析のためのプラットフォームの開発に携わっています。 実際、ニュースと興味深いソリューションが表示されたら、チームと私はここでデータ分析のプラットフォームについてお話します。

プロローグ

それでは始めましょう。 気流とは何ですか？ これは、作業プロセスの開発、計画、および監視のためのライブラリ（またはライブラリのセット ）です。 Airflowの主な機能：Pythonコードは、プロセスを記述する（開発する）ために使用されます。 これは、プロジェクトと開発を整理するための多くの利点を意味します。実際、（たとえば）ETLプロジェクトは単なるPythonプロジェクトであり、特定のインフラストラクチャ、チームサイズ、およびその他の要件を考慮して、好きなように整理できます。 道具として、すべてが簡単です。 たとえば、PyCharm + Gitを使用します。 それは美しく、とても快適です！

次に、Airflowの主要なエンティティを検討します。 それらの本質と目的を理解したら、プロセスのアーキテクチャを最適に編成します。 おそらく、主な本質は有向非巡回グラフ（以降、DAGと呼びます）です。

DAG

DAGは、特定のスケジュールで厳密に定義された順序で実行するタスクの一種のセマンティックな組み合わせです。 Airflowは、DAGやその他のエンティティを操作するための便利なWebベースのインターフェイスを提供します。

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

開発者は、DAGを設計するときに、DAG内のタスクが構築されるオペレーターのセットを作成します。 ここで、もう1つの重要なエンティティであるAirflow Operatorについて説明します。

オペレーター

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

• BashOperatorは、bashコマンドを実行するための演算子です。
• PythonOperatorは、Pythonコードを呼び出すための演算子です。
• EmailOperator-電子メールを送信するための演算子。
• HTTPOperatorは、http要求を処理するための演算子です。
• SqlOperator-SQLコードを実行するための演算子。
• センサー-イベント（適切な時間、必要なファイルの外観、データベース内の行、APIからの応答など）を待機するオペレーター。

より具体的な演算子があります：DockerOperator、HiveOperator、S3FileTransferOperator、PrestoToMysqlOperator、SlackOperator。

また、独自の特性に焦点を合わせてオペレーターを開発し、プロジェクトで使用することもできます。 たとえば、MongoDBToHiveViaHdfsTransfer、MongoDBからHiveにドキュメントをエクスポートするための演算子、ClickHouseを操作するためのいくつかの演算子CHLoadFromHiveOperatorおよびCHTableLoaderOperatorを作成しました。 実際、基本的な演算子に基づいて頻繁に使用されるコードがプロジェクトに表示されるとすぐに、それを新しい演算子にアセンブルする方法を考えることができます。 これにより、さらなる開発が簡単になり、プロジェクト内のオペレーターのライブラリーを補充します。

さらに、タスクのこれらすべてのインスタンスを実行する必要があります。次に、スケジューラに焦点を当てます。

プランナー

AirflowのタスクスケジューラはCelery上に構築されています。 Celeryは、キューとタスクの非同期および分散実行を整理できるPythonライブラリです。 気流側では、すべてのタスクがプールに分割されます。 プールは手動で作成されます。 原則として、それらの目標は、ソースでの作業の負荷を制限するか、DWH内のタスクを類型化することです。 プールは、Webインターフェースを介して管理できます。

各プールにはスロット数に制限があります。 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. 無料のセロリがある場合、タスクはそれに行きます。 タスクでプログラムした作業は、1つまたは別の演算子の使用を開始します。

簡単です。

スケジューラは、さまざまなすべての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コードであるため、たとえばシャーディングされたソースで作業するときにコードの量を減らすための非常に便利な方法があります。 3つの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リポジトリ内の100,500個のテーブルを思い出します。

モニタリング

リポジトリが単純なため、自分に都合の良いタスクを監視するプロセスを自分で構築できます。 Zeppelinのメモ帳を使用して、タスクのステータスを確認します。

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

エアフローコードが開いているため、Telegramにアラートを追加しました。 タスクの各作業インスタンスは、エラーが発生した場合、開発チームとサポートチーム全体で構成されるTelegramのグループにスパム送信します。

Airflowのタスクの全体像であるZeppelinを介して、Telegram（必要な場合）を介して迅速な応答を取得します。

合計

気流は主にオープンソースであり、それから奇跡を期待する必要はありません。 実用的なソリューションを構築するために時間と労力を費やす準備をしてください。 達成目標、私を信じて、それは価値がある。 開発速度、柔軟性、新しいプロセスの追加の容易さ-あなたはそれを好きになるでしょう。 もちろん、プロジェクトの組織、エアフロー自体の安定性に多くの注意を払う必要があります。奇跡はありません。

現在、毎日約6.5万のタスクを実行するエアフローがあります。 それらは性質がまったく異なります。 さまざまな非常に具体的なソースからメインDWHにデータを読み込むタスクがあります。メインDWHにウィンドウを計算するタスクがあります。高速DWHにデータを公開するタスクがあります。多くの異なるタスクがあります。 数字で言えば、これはDWH（Hadoop）内のさまざまな複雑さの230の ELTタスク、約250のソースデータベースです 。これは、DWHとELTでETLデータ処理に分かれている4人のETL開発者のチームです。 DWH内のデータ処理と、もちろんサービスインフラストラクチャを扱う別の管理者 。

今後の計画

プロセスの数は必然的に増加しており、エアフローインフラストラクチャに関して行う主なことはスケーリングです。 Airflowクラスターを構築し、Celeryワーカーにいくつかのレッグを割り当て、タスクスケジューリングプロセスとリポジトリで複製ヘッドを作成します。

エピローグ

もちろん、これは、エアフローについて伝えたいこととはほど遠いですが、主なポイントを強調しようとしました。 食欲は食事とともに来ますので、試してみてください。