Apache AirFlowをDocker環境で構築して簡単なジョブを作る - け日記
仕事でApache AirFlowを使う機会がありましたので、調査がてらに、Dockerで環境を構築し、簡単なジョブを定義します。 AirFlow AirFlowはジョブのスケジューリング・監視を、コード(主にPython)で定義・制御するためのプラットフォームです。ワークフロー、データフローの実装・運用に使われます。 github.com 類似のソフトウェアには、前回・前々回で紹介したLuigiや、Treasure DatasのDigDagなどがあります。 AirFlowではスケジューラも提供されているため、Luigiよりも広い範囲をカバーします ワークフローを、AirFlowではPython、DigDagでは.dagファイルで定義します Dockerで環境構築 ローカルに環境をインストールすることもできますが、Dockerでサクッと環境構築します。 以下のレポジトリからdocker-c
Google Cloud Composer FAQ
参考までに、アイオワ (us-central) の金額は下記。 Composer 本体Web $0.074 x 24時間 x 31日 x 2台 = $110.112Database $0.125 x 24時間 x 31日 x 2台 = $186Web and database storage $0.273 x 20GB = $5.46Network egress $0.156 x とりあえず0 = $0WorkerGCE インスタンス n1-standard-1 $24.2725 x 3ノード = $72.8175永続ディスク $0.040 x 100GB x 3 ノード = $12$110.112 + $186 + $5.46 + $72.8175 + $12 = $386.38 Q. 高くない? 何かいい方法はない? 高い！！！ 単体でも高いが、開発・ステージング・本番と 3環境作ると
AirflowでDAG実行時にGUI, CLI, REST APIからパラメータを渡す - valid,invalid
Airflow webserver GUIの ▶️ 再生アイコンからTrigger DAGをクリックすると事前にDAG定義の内容で[DAG]が実行される。 ただ、以下のようなユースケースのために、実行するDAG（以下、DAG Run）にパラメータを渡したいことがある。 バッチによってはパラメータ付きで手動実行したい JenkinsのParameterized Build的な機能がほしい 一時的なデータ投入・更新バッチのために毎回DAG定義を書くのではなく、汎用的なDAGを用意してパラメータで実行するコマンドを動的に切り替えできると便利 Conf option DAG RunにJSON形式でパラメーターを渡す Conf というオプションがあり、様々なインタフェースから渡せる。 個人的には「2. GUIからパラメータ付き実行」が欲しかったもの。 1. CLIからパラメータ付き実行 CLIでは-
Airflowでのジョブ作成方法の紹介
はじめに 環境 WSL2: Ver20 Helm: v3.6.1 Airflow: ver2 前回の記事 前回の記事では、Node.js以外でのジョブ実行の方法を紹介しました。今回はその続きで、一般的なジョブの作成方法の紹介をしようと思います。 記事の内容 Airflowでは、ジョブをpythonで記述でき、依存関係をDAGとしてかけます。今回は以下の二種類の紹介をしようと思います。 Taskflow APIを用いたDAGの書き方 動的にDAGを作成する方法 今回の記事で紹介したコードは以下のレポジトリにあります。 TL;DR Taskflow APIを用いたジョブの書き方は、従来のAirflowでのDAGより記述は楽になった。 Taskflow APIについて Taskflow APIとはAirflow 2.0から導入された新しいDAGの記述方法です。 @dagや@taskを用いてDAG
Airflowを使用してDWH向けデータパイプラインを作る - Qiita
前提 GCP内でのデータパイプラインで、DWHはBigQueryを使用 DWHバッチ処理のパターン バッチ処理ベースのDWHデータパイプラインは 複雑なことをしない限り以下のパターンでほとんど網羅できると思われる。 GCSからCSV等のファイルロード（DROP-CREATE-INSERT） GCSからCSV等のファイルを整形してロード（PREP-DROP-CREATE-INSERT） 最新マスタ等のデータ更新（TRUNCATE-INSERT） 蓄積テーブル、サマリのデータ更新（DELETE-INSERT） あとはデータパイプラインとは直接関係ないが、運用系として以下の処理パターンもある バッチ処理の開始と終了時のメール通知 GCS内でのファイル移動（退避） Airflowでのデータパイプラインの作成方法 Airflowはpythonを使用して、DAGと呼ばれるジョブの固まりを定義したファイ
コードリーディングでAirflowの仕組みを理解する
こんにちは。Nowcastでエンジニアをしている片山（@fozzhey）です。 NowcastではワークフローマネジメントツールとしてAirflowを採用しています。 Airflowは日本でもよく採用されており、実際に活用されている方も多いと思います。 しかし、Airflowを使うことはできるけど、実際にどういう仕組みで動いているのかは分からない方が多いのではないでしょうか？ せっかくPythonで書かれているのに、Airflowのコードを読まないのはもったいない！ ということで、この記事ではAirflowのコードリーディングを行いたいと思います。 なるべくコードやGithubのリンクを貼っていますが、手元のエディターでAirflowのリポジトリを開きながら読んでいただくとより理解が深まると思います。 コードリーディングの題材題材とするDAGとタスクこの記事ではAirflowが提供するサン
AirflowのTips 11選 | フューチャー技術ブログ
概要Airflow 利用にあたって知っておいたほうが良いんじゃないかなと思う情報をまとめました。いわゆるハマりどころです。 Airflow 自体の基本的な説明(用語,DAG 定義方法,etc..) は省略しています。 基礎参照先 公式 Tutorial Dockerfile バージョン Airflow 1.10.4 Postgres 10.7 Tips 11 選 1. 日時 2. リトライ 3. start_date の変更ができない 4. 開始日時を過去日時にすると過去ジョブが自動実行される 5. context の活用 6. Macros 7. Task 間で値を受け渡す方法 8. dags ディレクトリ以下のディレクトリの切り方 9. UI の表示が UTC 固定 10. Web UI 上からジョブのパラメータを渡せない 11. DAG/Task の同時実行数制御 1. 日時Airf
Airflowの処理の一部をdbtに移行しようとして断念した話 - Classi開発者ブログ
こんにちは、データプラットフォームチームでデータエンジニアをやっている滑川(@tomoyanamekawa)です。 以前紹介したデータ分析基盤であるソクラテスの改善のためにCloud Composer(Airflow)で行っている処理のdbtへの置き換えを検討しましたが、導入を見送りました。 調べてみてdbtに対するわかりみも深まったので、その供養のために検討内容を公開します。 同じように検討している方の参考になれば幸いです。 dbtとは DWH(Data Ware House)でのquery管理やデータの品質、データリネージの問題を解決してくれるツールです。 すでに先人たちがいろいろな記事を公開してくれているので、詳細は説明しませんがこちらの文がdbtをよく表しています。 ELTの「T」を担当するツール データの前処理における作業をELT（Extract、Load、Transform）と
1
リリース、障害情報などのサービスのお知らせ
最新の人気エントリーの配信
j次のブックマーク
k前のブックマーク
lあとで読む
eコメント一覧を開く
oページを開く
1 Meet Apache Airflow · Data Pipelines with Apache Airflow