SilosETLやBIを中心に興味のあるトピックを扱います。Airflowの連携に関する以下の記事について紹介したいと思います。www.xplenty.com前提条件と初期セットアップAirflowのセットアップの詳細はここでは省略します。今回は諸々の考慮が不要なDockerコンテナを使用します。Desktopが必要です。Pro/EnterpriseエディションでHyper-Vが必要となります。Toolboxをお使いください。Toolboxのインストール手順はこちらです。Walker氏によるブログ(こちら)を見るとなぜAirflowがジョブスケジューラに対する素晴らしい選択肢になるかが分かります。puckel/docker-airflowこれにより、Python(3.7-slim-stretch)公式イメージに基づいたAirflowでコンテナが読み込まれます。Hubウェブサイトをご覧ください。Dockerコンテナ内のホスト上に作成されたDAGにアクセスするには、VirtualBoxでフォルダ共有を有効にします。options.作成されたフォルダには、DAG用のdagサブフォルダが含まれると仮定します。私の場合はc:\GoogleDrive\docker\airflow\dags\です。ここで、コンテナ内で実行されるアプリケーションにアクセスするために、いくつかのポートをマップする必要があります。この例では、ポート8080でAirflowを実行すると想定しています。same:共有フォルダへの変更が適用されていることを確認するために、Dockerを再起動します。restart追加のpythonパッケージについては、すべてのパッケージをリストするrequirements.txtファイルを作成しました。私の場合はc:\GoogleDrive\docker\airflow\requirements.txtそこに置いたものは何でもentrypoint.shスクリプトによってインストールされます。installコマンド(--userオプション)を実行します。)から実行して、それを試してみましょう。webserverちなみにオプションのリストは以下です。executedもし、全てのセットアップが上手くいったらAirflowにアクセスします。http://localhost:8080/admin/フォルダのマッピングやインストールされたパッケージのマッピングを確認したい場合、以下のコマンドを実行してください。"bash"以下のような結果が返されます。Xplentyと統合するスケジューラーを配置したら、それを活用して、Xplentyで作成されたグラフの作成および統合ジョブのスケジューリングに使用します。はじめにXplentyコンテナ用にシンプルなWrapperを用意します。パッケージを実行するためには以下の機能をサポートしていることが重要です。参考に上記のプロセスをハンドルするためのPythonスクリプトを簡単に作成できます。変数:まずImportと共にスクリプトで使用される以下の変数を設定するところからはじめましょう。名前を設定しますapi_keyはアカウントにアクセスするのに使いますapi_urlは後ほど必要となる機能で使用される一般的なURLです。run_jobパッケージを実行するメインとなるFunctionの引数は「package_id」のみです。documentation.にしたがってセットされます。UrlにJobsメソッドが追加されます。get_available_clustersというFunctionはクラスターが利用可能かどうかをチェックする際に実行されます。'jobs'利用可能なクラスターが稼働中でない場合、create_clusterというFunctionがクラスターを作成するのに実行されます。get_available_cluster()スクリプトはクラスターが利用可能になるまで待つ必要があります。package_id=package_id)全て完了したらPostリクエスト全て完了したら、POSTリクエストがヘッダ情報、データおよび認証キーと一緒に先ほどのURLに送信されます。ジョブリクエスト送信後、スクリプトは結果のモニタリングを開始します。APIでジョブのステータスを30秒ごとに確認します。print(job_status)最後にジョブのステータスが返されます。以下は完全なFunctionです。job_statusget_available_clusters前述したように、このFunctionは全てのクラスターをフェッチし、それらのステータスをチェックします。FunctionはクラスターIDを返します。利用可能なクラスターがない場合は-1を返します。APIの詳細はこちら(here)で確認できます。"available"}'で利用可能な物だけをリストアップできます。以下のようにurlに追加されます。''))create_clustersちなみにXplentyのUI画面でクラスターをセットアップすることが可能です。SandboxクラスターかProductionクラスターか、ノードの設定、さらにクラスターがInactiveになってからどれくらいの時間間隔でクラスターをTerminateするかについて数回クリックするだけで作成可能です。今回のブログでは、以下のFunctionによりSandboxクラスターが自動的に作成されます。ヘッダー情報と今回使用するクラスターのパラメーターに関する情報を保持するdataという変数(ドキュメント(こちら)に従い、サンドボックスはシングルノードのみを使用する必要があります)と共にPOSTリクエストが送信されます。get_job_status最後のFunctionは実行中のジョブのステータスをフェッチするのに使われます。job_idパラメータを引数とし、適切なヘッダ情報と共にGETリクエストを実行しジョブのステータスが取得されます。DAGを作成しましょう!今回は、3つのパッケージを順番に実行するシンプルなDAGを作成します。それぞれジョブが完了するまで待機し、次のジョブをスタートします。imライブラリをインポートします。datetime次に、1日1回のスケジュールと、開始タスクとして使用される「dummy_operator」を含むDAG定義を作成します。*',オペレータ:今回は非常にシンプルです。各パッケージについてそれぞれ1つのオペレータを用意し、PackageIDの箇所が異なるだけで同一のFunctionを使用します。毎回基本的なPythonオペレータが使用されています。dag=dag)最後に依存関係を設定します。前述したとおり、ジョブは順次実行されるように設定します。Containerを起動し、DAGをチェックして実行し、Xplenty側で検証する必要があります。Containerを実行する以下のコマンドでコンテナを起動します。webserverその後コンテナが起動していることをログで確認します。私の環境では、optimistic_paniniというのがコンテナの名称なので以下のコマンドで確認します。optimistic_paniniログを確認すると、DAGが読み取られ、「SequentialExecutor」が使用されていることがわかります。/usr/local/airflow/dagshttp://localhost:8080/admin/へアクセスし、AirflowのUI画面が表示されることを確認します。DAGが有効であることかつスケジュールを確認します。Linksメニューの左端のボタンで直接DAGを実行することができます。DAGの名前をクリックすると、ジョブの実行結果などを確認できるようになります。複数の実行が列に表示され、統計が表示されます。失敗などのさまざまなステータスを表示しています。Xplentyではどのように反映されているかを確認してみましょう!!見ての通り、XplentyのジョブステータスはAirflowに反映されます-最新の実行(Airflowツリービューの右端の列)は次のことを示しています。salesfore_mysql_upsert_operatorは実行されませんでした。それはXplentyダッシュボード画面ではリストに表示されていません。s2_mysql_operatorは失敗しました。Xplentyダッシュボード画面の一番上でにFailedのステータスと共に表示されています。rest_s3_operatorは最初に実行され、成功しています。Xplentyダッシュボードでもリストの二番目で100%完了と表示されています。各実行ごとにGraphビューで以下のように確認することができます。ガントチャートでも確認できます。言った設定やログの確認を行うことができます。off!さて、すべてのセットアップが完了したら、実際に本番環境を作成し、企業全体に展開するのはどれくらい難しいのかと疑問に思うかもしれません。必要なすべてのインフラストラクチャ、サーバー構成、メンテナンスと可用性、ソフトウェアのインストールを考慮に入れると、スケジューラの信頼性を確保するために多くのことを実現する必要があります。誰かがこういった心配をすべて取り除いてくれて、ジョブのスケジュールに集中できるようにしたらどうでしょう?ということで、ここからは、Astronomer.ioとXplentyをさらに詳しく見ていきます。実際に彼らが主張しているように簡単かどうかを見てみましょう。手始めにHP(page)で公開されているガイドを見てみましょう。トライアルアカウントをセットアップしWorkspaceを作成します。Executorを選択します。Astronomer.ioの説明をここで紹介します。「Airflowは複数のexecutorプラグインをサポートしています。これらのプラグインは、タスクを実行する方法と場所を決定します。Executorをサポートしています。podですべてのタスクを起動します。AirflowをOpenする画面が表示されます。皆さんも気づかれたかもしれませんが、サーバーが裏で作成されます。生成されたウェブアドレスをみて気付くかもしれません。私の場合は以下のようになっています。環境全体がバックグラウンドで開始されるため、少し時間がかかる場合があります。起動したら(ブラウザウィンドウを更新して確認します)、以下のようにAirflowメイン画面が表示されます。画面にDAGがなく、完全に空の状態です。DAGをアップロードするUIオプションもありません。instructions.)に従う必要があります。Astronomer.io上でDAGを設定するWindows上でWSLを実行するHyper-Vを有効にしてWindowsバージョンを実行している場合にかぎり、WSLを使用して以下の手順を行ってください。以下の手順に従ってCLIをインストールしましょう。bashこれですべての設定が完了します。設定が正しく行われたかどうかを確認するには、astroコマンドを実行してヘルプが表示されるかどうかを確認してください。プロジェクトを作成し、作成したディレクトリに移動します。Cloudに接続することができます。gcp0001.us-east4.astronomer.io画面の指示に従ってログインします。oAuthを使用するか、ユーザー名/パスワードを使用します。完了したら確認メッセージが表示されます。APIキーを.envに入れてください。あなたのDAGをプロジェクトにコピーできるはずです。(私の環境では、/mnt/c/Astronomer/xplenty/dag)。deployコマンドを実行します。このコマンドは、最初にDeploymentとWorkspaceのオプションを提供します。今回の例では、リストに1つしかありません。最終的には果、セットアップ全体がAstronomer.ioにPublishされます。Workspaceに新しいDAGが利用可能になります。Docker上で実行する上記のように動くはずでしたが、実際には何も起こりませんでした。deploy」コマンドを実行すると以下のようにエラーにより実行できませんでした。1vic@DESKTOP-I5D2O6C:/c/Astronomer/xplenty$Hyper-Vがないため、私のWindowsでDockerデーモンが起動しません。Imageを読み込みます。CLIをインストールします-上記で実施したのと同じように。コンテナをインタラクティブモードで起動するには以下のコマンドを実行します。init」でプロジェクトを初期化し、確認メッセージを確認します。Cloudに接続することができます。gcp0001.us-east4.astronomer.io画面の指示に従ってログインします。oAuthを使用するか、ユーザー名/パスワードを使用します。Token:Tokenを取得して貼り付けます(うまく機能しました)もしくはユーザー名とパスワードを使用します。完了したら確認メッセージが表示されます。registry.gcp0001.us-east4.astronomer.io手順が完了したら、Dockerコンテナの状態を新しいイメージに保存しておくと便利です。psでコンテナIDを確認するだけです私の場合、270c02e5d9d5です。charming_galileo私は以下のコマンドでAstronomerがインストール済みのイメージを作ることができます。ubuntu:astroこれで新しいイメージが作成されました。imagesmコマンドで確認することができます。コンテナ内で作成したXplentyプロジェクトにコピーするDAGボリュームをマウントして、コンテナを再実行しました。私はdocker.sockをマウントすることで、astroがコンテナ内からdockerにアクセスできるようにしました。"bash"さて、デプロイの前の最後の手順はAPIキーです。次のように、Dockerfileの環境変数として設定することをお勧めします。xpl_api_key=<your-API-key-here>Deploymentのための設定が全て完了しました。deployを実行するだけです。succeeded!お気づきかもしれませんが、私の環境では既にいくつかのレイヤーが存在していました。一連のDeploymentの場合、一部の部品が再利用されていることがわかります。全ての設定が完了するのに数回の試行が必要でした。この例では問題が発生した場合に手順を複数回実施しました。実験することを恐れないでください!完了したら、Astronomerワークスペースに移動してDAGを実行できます。Wrapperによってクラスターが作成されました。ジョブが開始されました。すると、Xplentyですべてのジョブが完了したことを確認できます。グラフビューに反映されました。ツリービューで見るとおまけ好奇心が強い人は、途中でいくつかの失敗に気付くでしょう。特定の実行にマウスを合わせるだけで、詳細を取得できます。タスクの「State」は明らかに失敗しています。ボタンをクリックするだけで、ログを確認できるメニューが表示されます。ログを確認するのは本当に簡単なのですが、あまり細かく表現されていないのでわかりません。それでは、もう少し深く掘り下げて、Xplentyを調査してみましょう。details」をクリックするだけで簡単に確認できます。エラーおよびジョブで使用している変数を確認できるパネルが開きます。ジョブが失敗した要因が判明しました。"xplenty://XPLENTY_USER_S3_CONNECTION_9669@xplenty.dev/mako/breakfast.csv"ソースファイルが確認できなかったのが原因のようです。Xplentyで障害調査とエラーの要因分析がどれくらい簡単にできるかやってみました。AirflowとXplentyの統合AirflowとXplentyと一緒に使用することで、シームレスにジョブのスケジュールおよび監視を行うことができ、エンタープライズ規模のワークフローを実現することが可能になります。XplentyはクラウドベースのコードフリーのETLソフトウェアであり、シンプルな視覚化されたデータパイプラインを提供して、幅広いソースとデスティネーションをカバーする自動化されたデータフローを実現します。Athenaに関する記事をご紹介したいと思います。月別アーカイブはてなブログをはじめよう!datapipelineさんは、はてなブログを使っています。ブログを報告するスターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。