はじめに
はじめまして。
株式会社ビデオマーケットでデータエンジニアを担当しているT_Abeです。
弊社では、社内の権利料の集計処理を「Talend（タレンド）」というツールで行っています。
Talendは、GUIベースでローコードで処理の流れを組み立てられる便利なツールですが、長く使ううちにいくつかの課題が見えてきました。
- 新しく触る人にとって学習コストが高い
- どんな処理をしているのか中身が分かりづらい（ブラックボックス化）
- 当時の開発担当者が退職しており改修やトラブル対応が難しい
その結果、日々の集計業務でエラー対応に時間がかかることが増えていました。
こうした状況を改善するために、私たちのチームでTalendからAirflow（エアフロー）というワークフロー管理ツールへ移行することを決めました。
本記事では、この「TalendからAirflowへの移行」で直面した課題や工夫を紹介していきます。
これからAirflowを導入したい方や、既存システムの見直しを考えている方の参考になれば幸いです。
なお、本記事では、Apache Airflow 2.7系の環境を前提に構成・検証を行っています。
Talendについては、無償版のTalend Open Studio（Version 7.3.1）を利用していた環境をもとに記載しています。
記載している内容や機能は、これらのバージョンに基づいています。
- 1. Talendとは／Airflowとは
- 2. Talend運用時の課題
- 3.Airflowを選んだ理由
- 4. リプレイスの流れ
- 5.リプレイスで直面した課題
- 6. 工夫したこと
- 7. 現時点で得られた効果
- 8. まとめと今後の展望
1. Talendとは／Airflowとは
まず、今回のテーマである「Talend」と「Airflow」について簡単に紹介します。どちらもデータの集計や加工などを自動で処理するためのツールですが、仕組みや考え方が大きく異なります。
Talendとは
Talend（正式名称Talend Studio）とは、画面上で処理（ジョブ）の流れを組み立てられるツールです。
プログラミングの知識がなくても、パーツをドラッグ＆ドロップしてデータの流れを作れる点が特徴です。
一方で、処理の内容が複雑になるとどこで何をしているかが分かりづらくなるという課題があります。
Airflowとは
Airflow（正式名称Apache Airflow）とはプログラミング言語であるPythonを使って 「いつ・どの順番で・どんな処理を行うか」をコードで定義するツールです。
このコードの集合をDAG（Directed Acyclic Graph：有向非巡回グラフ）と呼び、処理の流れを明確に管理できます。
各処理（タスク）はOperator（オペレーター）という単位で定義され、実際の動作内容を指定します。
Airflowでは、処理内容をコードとして管理できるため、
- 変更点がGitで追跡できる
- 処理の意図を他のメンバーが理解しやすい
- トラブル発生時の修正や再実行がしやすい
といった利点があります。
Operatorとは
Airflowでは、各タスクを実行する最小単位をOperatorと呼びます。
たとえば上記の例で使われているPythonOperatorは、Python関数を実行するためのOperatorです。
他にも次のような種類があります。
| Operator名 | 役割 |
|---|---|
| PythonOperator | Python関数を実行 |
| BashOperator | シェルコマンドを実行 |
| BigQueryOperator | BigQueryのSQLを実行 |
| EmailOperator | メール送信を実行 |
OperatorはAirflowの中核的概念であり、「どんな処理をどのように実行するか」を定義します。
- DAGの一例
from datetime import datetime from airflow import DAG from airflow.operators.python import PythonOperator # タスクで実行する関数 def hello_world(): print("Hello, Airflow 2.7!") # DAG定義 with DAG( dag_id="simple_hello_world_dag", # DAGの識別子 start_date=datetime(2024, 1, 1), # 初回実行日 schedule_interval=None, # 手動実行のみ（スケジュールなし） catchup=False, # 過去分の実行をスキップ tags=["example"], # タグ（UIでのフィルタに利用可） ) as dag: # タスク定義 hello_task = PythonOperator( task_id="say_hello", python_callable=hello_world ) hello_task
- DAGの実行結果
Hello, Airflow 2.7!
2つのツールの違い
| 観点 | Talend | Airflow |
|---|---|---|
| 設計方法 | GUIで処理を構築 | コードで処理を定義 |
| 開発環境 | 専用ソフト(Talend Studio) | Python環境で開発 |
| 処理の見通し | 視覚的だが複雑化しやすい | コード化で構造が明確 |
| 管理方法 | バイナリ形式でGit管理が難しい | テキスト形式でGit管理が容易 |
Talendは視覚的に操作しやすいツール、Airflowはコードで柔軟に管理できるツールです。
それぞれに強みはありますが、データエンジニアリングとして「処理の中身をチーム全体で見える化し、改善していく」運用を目指す場合は、Airflowの仕組みがより適していました。
2. Talend運用時の課題
Talendは、GUI上で直感的に処理を組み立てられるという利点がある一方で、長期的な運用の中でいくつかの課題が明らかになってきました。
特に「チームでの開発・保守」という観点では、次のような問題が大きなネックになっていました。
開発環境の依存が強い
Talendはローカル環境に依存するため、個人の使用PCごとに環境構成や動作に差異が生じやすく、バージョン差異の対応にも時間がかかりました。
特に厄介だったのが、JDK(Java開発環境)のバージョン依存です。
Talend Open Studioが要求するJDK8と、Apple Silicon Macで利用できるJDKバージョンが合わず、動作しないケースが発生していました。
そのため、Apple Silicon MacではJDKの互換性問題を回避するために、Rosetta経由でx86_64版のJavaを導入したり、複数バージョンを切り替えながら検証を行う必要がありました。
こうした環境構築の再現性の低さや検証コストの高さは、結果的に「より構成管理しやすい仕組み(Airflow)」への移行を後押しする要因にもなりました。
担当者依存によるメンテナンス難
過去にジョブを作成したメンバーが退職しており、処理内容の意図や設計思想を正確に把握できる人がいない状態でした。
そのため、エラー発生時のリカバリ対応や仕様変更が属人的になり、修正にも時間を要するケースが多く見られました。
Git管理が困難(ソースがバイナリ形式)
Talend Open Studioのジョブファイルはバイナリ形式で保存されるため、Gitでの差分管理やレビューが困難でした。
結果として、変更履歴の追跡やレビュー体制の構築が難しく、チーム開発における透明性の確保に課題がありました。
GUIベースゆえの複雑化
GUIでの処理設計は視覚的にわかりやすい反面、複雑な条件分岐やデータ結合処理が増えると「どこで何をしているか」が見えにくくなるという問題があります。
ジョブ全体が肥大化し、改修時には依存関係の把握や影響範囲の確認に時間がかかっていました。
これらの課題は、属人化・ブラックボックス化・管理の難しさに直結しており、日常運用の安定性や改善速度に大きく影響していました。
こうした背景から、「より透明性があり、コードで管理できる仕組み」への移行が必要となりました。
3.Airflowを選んだ理由
TalendからAirflowへの移行を決めた背景には、Talend Open Studio(無償版)の提供終了という外的要因がありました。
2024年1月末をもって、OSS版であるTalend Open Studioの配布と更新が終了し、今後は新しいバージョンやサポートが提供されないことが公式に発表されています。
このため、既存ジョブを長期的に維持することが難しくなりました。
Talendには無償版(Talend Open Studio)と有償版(Talend Data Fabric / Qlik Talend Cloud)が存在し、有償版ではGitを用いたバージョン管理やチーム開発機能、ジョブスケジューリングなどが利用可能です。
一方で、チームで利用していた無償版ではこれらの機能が限定的で、ローカル環境で個別に構築・実行する運用が中心でした。
その結果、環境構築の再現性や処理内容の可視化に課題が生じていました。
また、有償版へ移行したとしてもGUIベースでの操作が中心である点は変わらず、処理のブラックボックス化やコードレビューの難しさといった構造的な課題は残ると判断しました。
これらを踏まえ、TalendからAirflowへの移行を決めた最大の理由は、「コードで処理を定義し、チーム全体で透明性をもって管理できる仕組み」を実現するためでした。
ここでは、Airflowを採用した主な理由を説明します。
コードベースでの管理が可能
Airflowはすべての処理をPythonコードで定義します。
そのため、処理内容の変更点をGit上で簡単に確認でき、レビューや履歴管理も通常のソフトウェア開発と同じプロセスで行えます。
これにより、
- 誰が・いつ・どのような変更を加えたかが明確になる
- 設計思想をドキュメント化せずにコードから理解できる
- 開発者間でのレビューや改善がスムーズ
といった開発におけるベストプラクティスを実現できました。
可視化と依存関係の明確化
Airflowでは、DAG(処理の流れ)を自動で可視化するWeb UIが提供されています。
どのタスクがどの順番で実行されるのか、どの処理が失敗しているのかを一目で確認でき、トラブル時の原因特定やリトライも容易です。
また、依存関係が明示的にコードで表現されるため、「何の後にどの処理を実行するのか」が明確になり、ジョブの設計段階から全体像を把握しやすくなりました。
運用面の強さ(リトライ・アラート・スケジューリング)
Airflowはスケジューラとしての柔軟性と運用制御の強さが特徴です。
- タスク失敗時の自動リトライ設定
- 処理異常を通知するSlackアラート
- 柔軟なスケジュール管理(cron形式・依存トリガー)
これらを組み合わせることで、人手に頼らない安定したバッチ運用を実現できました。
Google Cloudとの親和性
弊社では、日々の業務でBigQueryやCloud StorageなどのGoogle Cloudプロダクトを利用しています。
Airflowはこれらのサービスと高い親和性を持ち、BigQueryOperatorやGCSHook*1など、GCP専用コンポーネントが標準で用意されています。
さらに、Cloud Composer(AirflowベースのGCPサービス)を使うことで、
- Git管理したAirflowコードをそのままデプロイ可能
- インフラ管理を意識せずワークフロー実行
- GCPサービスと直接連携可能
という利点が得られました。
この「Cloud Composerとの統合による運用性とデプロイ容易性」も採用を後押しした重要なポイントでした。
4. リプレイスの流れ
TalendからAirflowへの移行は、単なるツール置き換えではなく、既存ジョブの構造を理解し直し、再設計するプロセスとして進めました。
ここでは、その実際の流れを紹介します。
①Talendジョブの構造を紐解く
まずは、既存のTalendジョブの中身を丁寧に読み解くことから始めました。
処理の流れ・参照テーブル・出力先など、現行仕様をすべて洗い出し、不要処理や不明点は関係者に確認して整理。
この段階で目的・依存関係を明確化することが、後のAirflow化をスムーズにしました。
②Airflow向けの仕様を定義
Talendジョブ全体像を把握した後、Airflowで再構築するための設計書を作成。
DAG分割単位、Operator切り出し、外部SQL呼び出し位置などを整理しました。
これは「TalendジョブをAirflowに置き換える設計図」としてまとめ、レビューや再利用を容易にしています。
③DAGへの落とし込み
定義した仕様をもとにAirflow上で実装。
処理をPythonコード化し、SQLや外部ファイルを分離して再利用性を高めました。
命名規則・タグ・スケジュールの統一により、チーム全体で保守しやすい構成を整備しました。
④検証(Talendの集計結果との突き合わせ)
Airflowで集計した結果をTalendの集計結果(正解データ)と突き合わせ、複数月分のデータを比較し、差分が意図的か・想定外かを確認しました。
マスタ更新や契約変更など集計時期による差分は許容範囲とし、ロジックの一致性を最重視しました。
その過程で、お客さまへ報告する数値の一部に1円単位の微差が発生しているケースを確認しました。
詳細に確認したところ、Talend側では計算結果が整数値に非常に近い場合に自動的に四捨五入される挙動があり、Airflow側では小数点以下を切り捨てていたことが原因でした。
たとえば、計算結果が 832.9999995円 となる場合、Talendでは833円に丸められ、Airflowでは832円として処理されていました。
この差異を踏まえ、集計ロジック内での丸め処理を明示的に統一し、Talendとの結果が完全に一致することを確認しました。
移行検証を通じて、従来処理の丸め仕様を再確認し、より正確な集計を行える体制を整えられたことも大きな収穫でした。
5.リプレイスで直面した課題
移行作業を進める中で、単純なツール置き換えでは解決できない課題がいくつか見えてきました。
ここでは、特に検討・調整に時間を要したポイントをまとめます。
TalendジョブをそのままDAGに置き換えられない
Talendで作られたジョブは、構造をそのまま移しても効率的に動作しませんでした。
業務フローやテーブル設計の見直しが必要で、設計思想の違いを理解することが重要でした。
設計思想の違い
Talend: ジョブ単位で設計(粒度細かく独立しやすい)
Airflow: 依存管理中心の設計(ジョブ統合・制御前提)
この違いにより、構成の整理・統合を同時に進める必要がありました。
DAG分割と命名規則の標準化
DAGの粒度・命名統一ルールの策定に時間を要しました。
業務単位＋共通タグ管理を原則とし、過剰分割や巨大DAGを防止しました。
Airflow固有機能の理解
XCom(タスク間データ連携)、Datasets(依存トリガー管理)、Variables(設定値の集中管理)、Connections(外部接続情報の一元管理)、TriggerRule(タスク実行条件制御)は、初期は学習コストがありました。
運用を重ねる中で、
- XCom: 軽量データの受け渡し
- Datasets: 依存DAGの自動トリガー
- Variables: 環境や設定値の切り替えを容易に管理
- Connections: 接続情報をコードから切り離して安全に運用
- TriggerRule: 失敗時や一部成功時など柔軟な制御を実現
といった特性を活かし、再利用性・保守性・可読性が向上しました。
Cloud Composer特有の制約と対応
移行中、ローカル環境では動いていたDAGがComposer上で動作しないという問題に直面しました。
発生した問題：
ローカルでは、1つの親DAGから子DAGを順次トリガー実行する構成を採用していましたが、Cloud Composer上ではこの構成がうまく動作せず、実行が途中で停止してしまいました。
このときTriggerDagRunOperator*2を利用していたため、複数のDAGトリガー処理がスケジューラに集中し、Composer環境ではパフォーマンス面（スケジューラ負荷やトリガー待機遅延）の影響を受けて安定して動作できなかったと考えられます。
対応策：
この問題は、Airflow 2.4で追加されたDatasets機能を活用することで解消しました。
親子関係を明示的にトリガー制御する代わりに、データセット更新をトリガーとする自然な依存構造に置き換えました。
結果的に構成がシンプルになり、実行順序が直感的に分かりやすいDAG設計に改善できました。
ゾンビタスクの多発と解消
Composer上での実行中に、タスクが終了せず「ゾンビタスク」として残る現象が頻発しました。
原因を調査した結果、ComposerインスタンスのCPUリソース不足が要因でした。
対応策としては、Composerのマシンタイプを一段階上げてCPU・メモリを増強。これによりゾンビタスクは発生しなくなり、安定したジョブ実行が実現しました。
この経験を通じて、DAG設計だけでなくComposerリソース設計も重要であることを学びました。
移行の過程では多くの課題がありましたが、結果的には、構造が整理され、運用しやすいAirflow基盤を構築することができました。
特に、Datasetsによる依存管理の導入とComposer環境の最適化が、システム全体の安定稼働に大きく寄与しました。
6. 工夫したこと
移行を進める中で、「どうすればAirflowをより扱いやすくできるか?」を意識し、チーム全体で改善を積み重ねました。
その中でも特に効果が大きかった工夫を紹介します。
DAGやSQLを外部ファイル化して再利用しやすくする
DAGファイルの中に直接SQLを書くのではなく、外部ファイルとして切り出すようにしました。
こうすることで、
- クエリの再利用がしやすくなり
- 処理の見通しが良くなり
- 他のDAGからも同じSQLを呼び出せるように
なりました。
DAG自体もすっきりして可読性が上がったことが大きなメリットです。
タグ機能でDAGを整理
Airflowのタグ機能を使って、DAGを業務・システムごとに分類しました。
これにより、Web UIでのフィルタリングが容易になり、「どの処理がどの業務に紐づいているか」をすぐに確認できるようになりました。
チーム内でも命名ルールを統一して、管理のしやすさを意識しています。
Datasets機能を使った依存管理
Airflow 2.4で追加されたDatasets機能を活用して、DAGの実行順を構成しました。
データの更新をトリガーとしてDAGを起動することで、
- 依存関係がコードで明確に
- 実行制御が柔軟に
- 全体構成がよりシンプルに
なりました。
これにより「どのデータ更新でどの処理が動くか」が可視化できるようになりました。
小さく分けて段階的に移行
最初からすべてのジョブを置き換えるのではなく、少しずつ検証しながら進める方式を採用しました。
1つのTalendジョブをAirflowのDAGに置き換えて動作確認 → 結果を比較 → 問題なければ次へ、という流れです。
このアプローチにより、リスクを抑えながら確実に移行できました。
ドキュメント化で属人化を防ぐ
開発と並行して、設計や運用ルールをドキュメント化しました。
命名規則、スケジュール設計、依存関係などを明文化し、「誰が見ても同じように開発・運用できる」状態を目指しました。
contextの活用とXComの使い分け
DAG内では、context（タスク実行時に自動で渡される情報）を使って、実行パラメータや日付などをタスク間でやり取りしています。
以前はXComを多用していましたが、現在は「軽いデータのみXCom、メインの情報はcontext」というルールに統一しました。
この設計に変えたことで、処理の責務が明確になり、DAGの再利用や保守が格段にやりやすくなりました。
7. 現時点で得られた効果
段階的な移行と設計の見直しにより、主に以下の観点でチーム全体の開発・運用効率が大きく改善しました。
- 処理の見える化でトラブルシュートが迅速化
- 全体の処理時間が短縮（冗長処理の整理・再利用）
- DAG設計の統一化でチーム理解が深化
- Talend時代の潜在バグを検出・修正
- 不要仕様や重複処理の削減でシンプル化
また、移行の過程で特に効果が大きかったのが、データ品質と一貫性の向上です。
Talendで生成されたデータには、一部で不整合や重複レコードが見られました。
Airflow化を機に、次のような取り組みを行いました。
- データ仕様の明確化（各テーブル・カラムの役割を定義）
- マスタ・参照定義の再整理（参照整合性を確保）
- 検証ロジックをDAGに組み込み、異常検知を自動化
これにより、集計処理だけでなくデータの品質保証プロセスもAirflow上で一元管理できるようになり、「正しいデータが、正しい手順で、正しいタイミングに生成される」体制を実現できました。
結果として、Airflowへの移行を通じて、「見える・直せる・改善できる、かつ品質を担保できる基盤」へと改善しました。
8. まとめと今後の展望
TalendからAirflowへの移行は、単なるツールの置き換えではなく、業務フロー全体を見直すきっかけになりました。
今後は、
- DAGの自動生成やスケジュールの最適化
- CI/CDによる自動デプロイ
- モニタリングや通知の強化
など、さらに運用の改善と自動化を進めていきます。
この移行を通じて、データ基盤の安定性とスケーラビリティの向上を目指していきます。
