Posted inデータベース

AirbyteをDockerにデプロイする：データベースとウェアハウス間のELTデータ同期を自動化

Database tutorial - IT technology blog
Database tutorial - IT technology blog

あなたも直面したことがあるデータ同期の問題

PostgreSQLの本番データベース、MongoDBの分析ストア、Redshiftウェアハウス、さらにサードパーティのSaaSなど、複数のシステムをまたいでデータを扱っていると、保守の悪夢はあっという間にやってくる。最初はcronジョブとPythonスクリプトを数本書くだけで済む。ところが半年後には、誰も全容を把握していないスクリプトが14本に増え、その半分はサイレントに失敗し、データチームはデータ分析よりパイプラインの火消しに費やす時間の方が多くなっている。

原因は怠慢ではない。手書きのETLはそもそもスケールしないのだ。新しいデータソースが増えるたびに、新しい連携、新しい認証処理、新しいエラー回復ロジック、新しいモニタリングが必要になる。ソースとデスティネーションの組み合わせが爆発的に増えていく中で、多くのチームは最終的に行き詰まってしまう。

そこで登場するのがAirbyteだ。300以上のビルド済みコネクタと標準化された同期プロトコルを備えており、接続を一度設定すれば、認証・リトライ・スキーマの変更はプラットフォームが自動で処理してくれる。ELTアプローチでは、まず生データをデスティネーションにロードし、その後dbtやSQLを使ってウェアハウス内で変換を行う。ソースデータベースが変換の負荷を受けることはない。

なぜクラウドではなくDockerでAirbyteを動かすのか

このプラットフォームはDockerコンテナのセットとして提供されている。VPSやローカルマシンでDocker Composeを使って起動すると、次のメリットが得られる：

  • 完全なコントロール — データがインフラ外に出ることがない
  • セルフホスト運用ではSaaSのサブスクリプション費用が不要
  • docker compose pullで簡単にアップグレード可能
  • ローカルと本番で同一の環境を維持できる

フルマネージドの環境を求めるならAirbyte Cloudで十分だ。しかし、HIPAAの要件、社内のセキュリティポリシー、あるいは強いこだわりでデータをインフラ外に出せない場合、セルフホスティングが唯一の現実的な選択肢となる。データは自分のサーバーに留まる、それだけだ。

ELTが従来のETLと何が違うのかは少し立ち止まって考える価値がある。従来のETLはデータをロードする前に変換する — ビジネスルールが変わるたびにパイプラインを触る必要がある。ELTはこれを逆転させる。まず生データをウェアハウスにロードし、その後dbtモデルやSQLビューで変換を行う。ビジネスルールが変わったら？モデルを更新すればいい、同期処理は触らなくていい。インジェストレイヤーは安定したまま保てる。

インストール：Docker ComposeでAirbyteをデプロイする

前提条件

始める前に、以下を確認しておこう：

  • Docker Engine 20.10以上とDocker Compose v2
  • 最低4GBのRAM（複数の同時同期を行う場合は8GB推奨）
  • ポート8000（UI）と8001（API）が使用可能であること

まずバージョンを確認しよう：

docker --version
docker compose version

Airbyteのプルと起動

Composeファイルを自分で書く必要はない — Airbyteには全セットアップを処理する公式の起動スクリプトが付属している。3つのコマンドで起動できる：

# 作業ディレクトリを作成する
mkdir airbyte && cd airbyte

# 公式の起動スクリプトをダウンロードする
curl -LsfS https://raw.githubusercontent.com/airbytehq/airbyte/refs/heads/master/run-ab-platform.sh \
  -o run-ab-platform.sh

# 実行権限を付与してバックグラウンドで起動する
chmod +x run-ab-platform.sh
./run-ab-platform.sh -b

-bフラグはすべてをバックグラウンドで実行する。初回起動時はすべてのイメージをプルするため、接続速度によって5〜10分かかる場合がある。

起動したらhttp://localhost:8000を開こう。デフォルトの認証情報はairbyte / passwordだ。

リモートサーバーで動かす場合は、起動前に認証情報を変更すること。Airbyteが生成する.envファイルで設定できる：

# 初回起動前にairbyte/.envを編集する
BASIC_AUTH_USERNAME=yourname
BASIC_AUTH_PASSWORD=a-strong-random-password

すべてのコンテナが正常に起動したことを確認しよう：

docker compose ps

airbyte-server、airbyte-webapp、airbyte-worker、airbyte-db、airbyte-temporalのすべてがrunning状態になっているはずだ。再起動ループに入っているコンテナがあれば、ログを確認しよう：

docker compose logs <container-name>

設定：最初のELTパイプラインを構築する

ソースの追加（PostgreSQLの例）

Airbyteの管理画面でSources → New Sourceに移動する。「Postgres」を検索し、接続情報を入力しよう：

  • Host：データベースのホスト名またはIPアドレス
  • Port：5432
  • Database：myapp_production
  • Username / Password：データベースの認証情報
  • Replication Method：Standard（リアルタイムの変更キャプチャにはCDC）

注意点がある：同じマシン上のDockerでPostgresを動かしている場合、ホストにlocalhostを使ってはいけない。そのアドレスはホストマシンではなくAirbyteコンテナ内で解決されてしまう。共有Dockerネットワークで解決しよう：

docker network create airbyte_network

AirbyteのComposeファイルとデータベースのComposeファイルの両方のnetworks:に、external: trueとして追加する。そうすればAirbyte上のホストにはPostgresのコンテナ名を使える。

デスティネーションの追加

Destinations → New Destinationに移動する。BigQuery、Snowflake、Redshift、別のPostgresインスタンス、S3など多数の選択肢がある。クラウドアカウントなしでローカルテストをする場合は、Local JSONデスティネーションを使おう — 同期されたデータをAirbyteコンテナ内の/tmp/airbyte_local/にJSONファイルとして書き出す。

サンプルデータでテストする場合、CSVエクスポートをまずJSONに変換する必要があることが多い。その際、toolcraft.app/ja/tools/data/csv-to-jsonはブラウザ上で完結して変換できる — アップロード不要、サーバー処理なし。サンプルデータに機密情報が含まれていてマシン外に出したくない場合に便利だ。

接続の作成と同期モードの選択

ソースとデスティネーションの準備ができたら、Connections → New Connectionに移動し、ソースとデスティネーションを選択して、ストリーム（テーブル）ごとの同期設定を選ぼう：

  • Full Refresh | Overwrite — 同期のたびにデスティネーションのテーブルを置き換える
  • Full Refresh | Append — すべての履歴レコードを保持する
  • Incremental | Append — 新規・変更された行のみ同期する（updated_atのようなカーソルフィールドが必要）
  • Incremental | Append + Deduped — スマートなupsert；本番OLTPからウェアハウスへのパイプラインに最適

ほとんどの本番環境では、Incremental | Append + Dedupedが正しいデフォルトだ。前回の同期以降に変更された行のみを転送し、ソース側のフルテーブルスキャンをスキップし、重複を蓄積することなくデスティネーションを正確に最新状態に保つ。

同期頻度（手動、毎時、6時間ごと、毎日）を設定し、同期するストリームを選択したらSet up connectionをクリックする。

検証とモニタリング

同期のトリガーと監視

接続を作成したら、UIのSync nowボタンから手動で同期をトリガーするか、ポート8001のAirbyte APIを使おう：

curl -X POST http://localhost:8001/api/v1/connections/sync \
  -H "Content-Type: application/json" \
  -d '{"connectionId": "your-connection-id"}'

接続IDは、管理画面で接続を表示したときのURL（/connections/<uuid>）で確認できる。

Job Historyを開いてジョブをクリックすると、ストリームごとの統計（同期された行数、転送バイト数、エラー時は完全なエラートレース）が確認できる。重要な挙動として、Airbyteはストリームレベルで失敗を分離する。1つのテーブルがエラーになっても、その同期の残りのストリームは完了まで実行される。

コンテナレベルのログ確認

# Workerが実際の同期ジョブを処理する
docker compose logs -f airbyte-worker

# ServerがAPIコールとスケジューリングを処理する
docker compose logs -f airbyte-server

PythonによるSync Statusのポーリング

同期ステータスを自前のモニタリングスタックに組み込みたい場合、Airbyte APIを使えば簡単に実現できる：

import requests

BASE = "http://localhost:8001/api/v1"
headers = {"Content-Type": "application/json"}

# 接続の最近の同期ジョブを一覧表示する
response = requests.post(
    f"{BASE}/jobs/list",
    json={"configId": "your-connection-id", "configType": "sync"},
    headers=headers
)

jobs = response.json().get("jobs", [])
for job in jobs[:5]:
    info = job["job"]
    print(f"ジョブ {info['id']}: {info['status']} | 作成日時: {info['createdAt']}")

同期失敗時のWebhook通知

本番環境では、Settings → Notificationsで失敗アラートを設定しよう。AirbyteはSlack Webhookをネイティブでサポートしている。SlackチャンネルやWebhook-to-PagerDutyブリッジに向けておけば、UIを手動でポーリングすることなく同期失敗の通知を受け取れる。

Airbyteのアップグレード

接続設定と同期履歴はairbyte_db Dockerボリュームに保持されるため、アップグレードしても設定が消えることはない：

# 最新イメージをプルして再起動する
./run-ab-platform.sh -b

# またはComposeで手動実行
docker compose pull
docker compose up -d

アップグレード前にAirbyteのリリースページを確認し、コネクタバージョンやAPI動作の破壊的変更がないかチェックしよう。ステージング環境で先にアップグレードを試すのは、10分余分にかかっても価値がある。

デスティネーションに行が届き始めれば、初期セットアップの壁は乗り越えたも同然だ。2つ目・3つ目のソースを追加する手順はまったく同じ：コネクタを選び、認証情報を設定し、同期モードを選ぶ。統合の複雑さはコネクタが吸収してくれる — あとは本来のデータ作業に集中しよう。

Share:
Tags: