Zenn
🔗

AirflowからAirbyteをトリッガーする際にハマるポイント

2023/03/06に公開
Docker
#
data
Airflow
#
データ基盤
Airbyte
tech

https://docs.airbyte.com/operator-guides/using-the-airflow-airbyte-operator/

AirflowからAirbyte Operatorを利用するための設定について、Airbyte公式の記事は既にわかりやすくまとめています。実際に試してみて、少しハマったところがあったので、その知見を共有したいと思います。

1. Airflowを2.3.0以上にアップグレードする必要がある

apache-airflow-providers-airbyte[http]を利用するのにAirflowを2.3.0以上に上げないといけません。(apache-airflow-providers-airbyte[http]をdocker-composer.ymlの_PIP_ADDITIONAL_REQUIREMENTSに追加することも忘れずに)
Cloud Composerなどを利用している場合、GUIからアップグレード可能です。

https://airflow.apache.org/docs/apache-airflow-providers-airbyte/stable/index.html

version: '3'
x-airflow-common:
  &airflow-common
  image: apache/airflow:2.3.4-python3.8
  environment:
    &airflow-common-env
    PYTHONPATH: /opt/airflow/dags
    AIRFLOW__CORE__EXECUTOR: CeleryExecutor
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:password@postgres/airflow
    AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:password@postgres/airflow
    AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
    AIRFLOW__API__AUTH_BACKEND: 'airflow.api.auth.backend.basic_auth'
    # 追加
    _PIP_ADDITIONAL_REQUIREMENTS: apache-airflow-providers-airbyte[http]==3.2.0

2. Airflowの古いバージョンから2.3.4上げるとdocker-composeがバグる

airflow 2.2.xでは問題なく環境構築できていましたが、イメージをapache/airflow:2.3.4-python3.8に変更してdocker compose up airflow-initを実行したら怒られます。

You are running pip as root. Please use 'airflow' user to run pip!

Airflowの古いdocker-composer.ymlのバグのようなので、
https://github.com/apache/airflow/pull/23517/files

services -> airflow-init -> environmentに_PIP_ADDITIONAL_REQUIREMENTS: ''を追加すれば解決できます。

...
    environment:
      <<: *airflow-common-env
      _AIRFLOW_DB_UPGRADE: 'true'
      _AIRFLOW_WWW_USER_CREATE: 'true'
      _AIRFLOW_WWW_USER_USERNAME: ${_AIRFLOW_WWW_USER_USERNAME:-airflow}
      _AIRFLOW_WWW_USER_PASSWORD: ${_AIRFLOW_WWW_USER_PASSWORD:-password}
      # 追加
      _PIP_ADDITIONAL_REQUIREMENTS: ''
...

3. connectionsを設定する際にCould not resolve host: host.docker.internalエラー

公式記事を参考にして環境構築し、airbyteのconnections設定してTestボタンをポチるとCould not resolve host: host.docker.internalというエラーが出ました。
調べてみたら、Docker Engine（Linux）でコンテナからホスト側のサービスにアクセスする（host.docker.internal）際にhost.docker.internal:host-gatewayをextra_hostsに追加する必要があります。

https://kazuhira-r.hatenablog.com/entry/2022/05/21/152825

version: '3'
x-airflow-common:
  &airflow-common
  image: apache/airflow:2.3.4-python3.8
  # 追加
  extra_hosts:
    - "host.docker.internal:host-gateway"
  environment:
    &airflow-common-env
    PYTHONPATH: /opt/airflow/dags
    AIRFLOW__CORE__EXECUTOR: CeleryExecutor
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:password@postgres/airflow
    AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:password@postgres/airflow
    AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
    AIRFLOW__API__AUTH_BACKEND: 'airflow.api.auth.backend.basic_auth'
    _PIP_ADDITIONAL_REQUIREMENTS: apache-airflow-providers-airbyte[http]==3.2.0

また、AWSやGCPのconnections設定と同じく毎回手動で設定するのがめんどくさいので、コンテナを立ち上げた後docker exec -it colossus-workflow-webserver airflow connections import /opt/airflow/config/connections.jsonを使ってconnectionsをimportすれば便利です。

{
  "airbyte_conn": {
    "conn_type": "airbyte",
    "host": "host.docker.internal",
    "login": "airbyte",
    "password": "password",
    "port": 8000
  }
}

以上3点をクリアして問題なくAirflowからAirbyteをトリッガーできました。
ご参考になれば幸いです。
AirflowやAirbyteなどの知見があればまた共有したいと思います。

Discussion