airbyte-connection-setup

Airbyte 接続設定エキスパート

Airbyte の接続設定と管理に関するエキスパートです。

基本原則

接続アーキテクチャ

• ソースと宛先のペア：変換を伴うデータフロー
• 増分同期：フルリフレッシュよりも優先されます
• スキーマの進化：スキーマの変更に対応します
• 障害からの復旧：エラーを確実に処理します

同期モード

• Full Refresh：データを完全に上書きします
• Incremental Append：新しいレコードを追加します
• Incremental Deduped：プライマリキーに基づいて重複を排除します

Docker Compose の設定

version: '3.8'
services:
  db:
    image: airbyte/db:${VERSION}
    environment:
      - POSTGRES_USER=${DATABASE_USER}
      - POSTGRES_PASSWORD=${DATABASE_PASSWORD}
    volumes:
      - db:/var/lib/postgresql/data

  server:
    image: airbyte/server:${VERSION}
    environment:
      - DATABASE_PASSWORD=${DATABASE_PASSWORD}
      - DATABASE_URL=${DATABASE_URL}
      - WORKSPACE_ROOT=/tmp/workspace
    ports:
      - "8001:8001"
    volumes:
      - workspace:/tmp/workspace
      - data:/data
    depends_on:
      - db

  webapp:
    image: airbyte/webapp:${VERSION}
    ports:
      - "8000:80"
    depends_on:
      - server

ソースの設定

# PostgreSQL источник
source_config = {
    "host": "localhost",
    "port": 5432,
    "database": "production_db",
    "username": "airbyte_user",
    "password": "secure_password",
    "ssl_mode": {"mode": "require"},
    "replication_method": {
        "method": "CDC",
        "plugin": "pgoutput",
        "initial_waiting_seconds": 300
    }
}

宛先の設定

# Snowflake назначение
destination_config = {
    "host": "account.snowflakecomputing.com",
    "role": "AIRBYTE_ROLE",
    "warehouse": "AIRBYTE_WAREHOUSE",
    "database": "AIRBYTE_DATABASE",
    "schema": "RAW_DATA",
    "username": "airbyte_user",
    "password": "secure_password",
    "loading_method": {"method": "Internal Staging"}
}

同期の設定

{
  "syncCatalog": {
    "streams": [
      {
        "stream": {
          "name": "users",
          "supportedSyncModes": ["full_refresh", "incremental"]
        },
        "config": {
          "syncMode": "incremental",
          "cursorField": ["updated_at"],
          "destinationSyncMode": "append_dedup",
          "primaryKey": [["id"]]
        }
      }
    ]
  },
  "schedule": {
    "units": 1,
    "timeUnit": "hours"
  }
}

API 統合

import requests

# Создание источника
source_payload = {
    "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",
    "connectionConfiguration": source_config,
    "workspaceId": workspace_id,
    "name": "Production PostgreSQL"
}

response = requests.post(
    f"{airbyte_url}/api/v1/sources/create",
    json=source_payload,
    headers={"Content-Type": "application/json"}
)

# Создание подключения
connection_payload = {
    "sourceId": source_id,
    "destinationId": destination_id,
    "syncCatalog": sync_catalog,
    "schedule": {"units": 1, "timeUnit": "hours"}
}

requests.post(
    f"{airbyte_url}/api/v1/connections/create",
    json=connection_payload
)

トラブルシューティング

よくある問題

1. ネットワーク接続：ファイアウォールとアクセスを確認します
2. 認証：権限と有効期限を確認します
3. リソース制限：メモリと CPU を監視します
4. スキーマの変更：設定を更新します

最適化

• バッチサイズを調整します
• 接続プーリングを使用します
• カーソルフィールドにインデックスを作成します
• パフォーマンスを監視します

セキュリティ

• シークレットを vault に保存します
• すべての接続で SSL/TLS を使用します
• ネットワークをセグメント化します
• 定期的に更新します