Airflow2.2.3 + Celery + MySQL 8 は堅牢な分散スケジューリング クラスタを構築します

先ほど、Airflow インフラストラクチャと、コンテナに Airflow をデプロイする方法について説明しました。今日は、Airflow と Celery を使用して堅牢な分散スケジューリング クラスターを構築する方法を見ていきます。

1 クラスタ環境

Ubuntu 20.04.3 LTS マシンにも Airflow クラスターをインストールします。今回はテスト用に同じ構成のサーバーを3台用意します。前回の記事[1]では、Bigdata1サーバーにairflowのすべてのコンポーネントをインストールしました。まだ読んでいない方は、リンクをクリックして、まずは前回の記事を読んでみてください。ここで、他の 2 つのノードにワーカー コンポーネントをインストールするだけです。

前回の記事の docker-compose.yml では、デプロイメント ファイルとデータ ディレクトリが分離されておらず、後からの管理に不便です。したがって、サービスを停止し、データベースとデータ ディレクトリをデプロイメント ファイルから分離することができます。

• デプロイメントファイル: docker-compose.yaml/.env は /apps/airflow ディレクトリに保存されます
• MySQLと設定ファイル: /data/mysql に配置
• エアフローデータディレクトリ: /data/airflow に配置

この分割により、後で統一された管理が容易になります。

2. ワーカーサービスをデプロイする

事前準備

mkdir /data/airflow/{dags,plugins} -pv
 mkdir -pv /apps/airflow
 mkdir -pv /logs/エアフロー

ワーカーデプロイメントファイル:

 ---  
バージョン: '3'  
 x-エアフロー共通:
  エアフロー共通
  ＃で 注文 に カスタム依存関係を追加したり、プロバイダー パッケージをアップグレードしたりすると、拡張イメージを使用できるようになります。
  # イメージ行をコメントアウトし、 docker-compose.yaml を配置したディレクトリにDockerfileを配置します。
  # 以下の「build」行のコメントを解除し、 ` docker-compose build` を実行してイメージをビルドします。
  イメージ: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.2.3}
  ＃ 建てる： 。
  環境：
    &エアフロー共通環境
    AIRFLOW__CORE__EXECUTOR: セロリエグゼキューター
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: mysql+mysqldb://airflow:aaaa@$${MYSQL_HOST}:3306/airflow #MySQLに対応するアカウントとパスワードを変更します
    AIRFLOW__CELERY__RESULT_BACKEND: db+mysql://airflow:aaaa@$${MYSQL_HOST}:3306/airflow #MySQLに対応するアカウントとパスワードを変更します
    AIRFLOW__CELERY__BROKER_URL: redis://:xxxx@$${REDIS_HOST}:7480/0 #Redisのパスワードを変更する
    AIRFLOW__CORE__FERNET_KEY: ''  
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'  
    AIRFLOW__CORE__LOAD_EXAMPLES: 'true'  
    AIRFLOW__API__AUTH_BACKEND: 'airflow.api.auth.backend.basic_auth'  
    _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:-}
  ボリューム:
    - /data/airflow/dags:/opt/airflow/dags
    - /logs/エアフロー:/opt/エアフロー/ログ
    -/data/airflow/plugins:/opt/airflow/plugins
    - /data/airflow/airflow.cfg:/opt/airflow/airflow.cfg
ユーザー: "${AIRFLOW_UID:-50000}:0"   
 
サービス:
  エアフローワーカー:
    <<: *エアフロー共通
    コマンド: セロリワーカー
    ヘルスチェック:
      テスト：
        - 「CMD-SHELL」  
        - 'celery --app airflow.executors.celery_executor.app 検査 ping -d "celery@$${HOSTNAME}"'  
      間隔: 10秒
      タイムアウト: 10秒
      再試行: 5
    環境：
      <<: *エアフロー共通環境
      #セロリワーカーのウォームシャットダウンを適切に処理するために必要
      # https://airflow.apache.org/docs/docker-stack/entrypoint.html#signal-propagation を参照してください
      DUMB_INIT_SETSID: "0"  
    再起動: 常に
    ホスト名: bigdata-20-194 # ここでコンテナのホスト名を設定すると、どのワーカーであるかを簡単に確認できます
    依存:
      エアフロー初期化:
        条件: サービスが正常に完了しました
 
  エアフロー初期化:
    <<: *エアフロー共通
    エントリポイント: /bin/bash
    #yamllintルールを無効にする:line-length
    指示：
      - -c
      - |
関数ver() {
          printf "%04d%04d%04d%04d" $${1//./ }
        }
        airflow_version=$$(gosu エアフロー エアフロー バージョン)
        airflow_version_comparable=$$(バージョン $${airflow_version})
        最小エアフローバージョン=2.2.0
        min_airflow_version_comparable=$$(バージョン $${min_airflow_version})
        比較可能なエアフローバージョンが最小エアフローバージョンの場合、それから 
          エコー
          echo -e "\033[1;31mエラー!!!: Airflowのバージョンが古すぎます $${airflow_version}!\e[0m"  
          echo "サポートされている最小の Airflow バージョン: $${min_airflow_version}。これ以降のバージョンのみを使用してください!"  
          エコー
          出口1
        フィ
        [[ -z "${AIRFLOW_UID}" ]]の場合;それから 
          エコー
          echo -e "\033[1;33m警告!!!: AIRFLOW_UIDが設定されていません!\e[0m"  
          echo "Linux を使用している場合は、以下の手順に従って設定する必要があります"  
          echo "AIRFLOW_UID 環境変数。そうでない場合、ファイルは root によって所有されます。"  
          echo "他のオペレーティング システムの場合は、手動で作成した .env ファイルで警告を取り除くことができます:"  
          echo "参照: https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html#setting-the-right-airflow-user"  
          エコー
        フィ
        1048576 メガバイト
        mem_available=$$(($$(getconf _PHYS_PAGES) * $$(getconf PAGE_SIZE) / one_meg))
        cpus_available=$$(grep -cE 'cpu[0-9]+' /proc/stat)
        disk_available=$$(df / | tail -1 | awk '{print $$4}' )
        warning_resources = "偽"  
        mem_available が 4000 未満の場合それから 
          エコー
          echo -e "\033[1;33m警告!!!: Dockerに使用できるメモリが不足しています。\e[0m"  
          echo "少なくとも 4GB のメモリが必要です。$$(numfmt --to iec $$((mem_available * one_meg))) があります"  
          エコー
          warning_resources= "true"  
        フィ
        利用可能なCPU数が2以下の場合それから 
          エコー
          echo -e "\033[1;33m警告!!!: Dockerに使用できるCPUが足りません。\e[0m"  
          echo "少なくとも 2 つの CPU を推奨します。$${cpus_available} があります"  
          エコー
          warning_resources= "true"  
        フィ
        ディスクの空き容量が 1 メガバイト未満の場合、それから 
          エコー
          echo -e "\033[1;33m警告!!!: Docker に使用できるディスク容量が足りません。\e[0m"  
          echo "少なくとも 10 GB を推奨します。$$(numfmt --to iec $$((disk_available * 1024 ))) があります"  
          エコー
          warning_resources= "true"  
        フィ
        [[ $${warning_resources} == "true" ]] の場合;それから 
          エコー
          echo -e "\033[1;33m警告!!!: Airflow を実行するためのリソースが不足しています (上記を参照)!\e[0m"  
          echo "利用可能なリソースの量を増やすには、次の手順に従ってください:"  
          「https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html#before-you-begin」をエコーし​​ます 
          エコー
        フィ
        mkdir -p /sources/logs /sources/dags /sources/plugins
        chown -R "${AIRFLOW_UID}:0" /sources/{logs,dags,plugins}
 exec /entrypoint エアフローバージョン
    #yamllintルールを有効にする:line-length
    環境：
      <<: *エアフロー共通環境
      _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:-airflow}
ユーザー: "0:0"  
    ボリューム:
      - .:/ソース
 
  エアフロー CLI:
    <<: *エアフロー共通
    プロフィール:
      - デバッグ
    環境：
      <<: *エアフロー共通環境
      CONNECTION_CHECK_MAX_COUNT: "0"  
    # エントリポイントの問題に対する回避策。参照: https://github.com/apache/airflow/issues/16252
    指示：
      -バッシュ
      - -c
      - 気流

テストを初期化し、環境が次の要件を満たしているかどうかを確認します。

 cd /apps/ariflow/
 echo -e "AIRFLOW_UID=$(id -u)" > .env # AIRFLOW_UID は通常のユーザーの UID である必要があり、これらの永続的なディレクトリを作成する権限を持っている必要があることに注意してください。
 docker-compose を起動し、airflow-init を実行します。

データベースがすでに存在する場合、初期化テストは既存のデータベースに影響しません。次に、airflow-worker サービスを実行します。

 docker-compose を起動 -d

次に、同じ方法でbigdata3ノードにairflow-workerサービスをインストールします。デプロイメントが完了したら、flower を通じてブローカーのステータスを確認できます。

3永続的な設定ファイル

ほとんどの場合、複数の Airflow ワーカー ノードを持つクラスターを使用する場合は、Airflow 構成ファイルを永続化し、すべてのノードに Airflow を同期する必要があります。そのため、docker-compose.yaml 内の x-airflow-common のボリュームを変更し、airflow.cfg をボリュームとしてコンテナにマウントする必要があります。構成ファイルをコンテナにコピーして変更することができます。

使用の初期段階では、docker-compose ファイル内のいくつかの環境変数の値を airflow.cfg ファイルに書き込む必要があります。たとえば、次の情報です。

 [コア]
 dags_folder = /opt/airflow/dags
ホスト名_callable = socket.getfqdn
 default_timezone = Asia/Shanghai # タイムゾーンを変更する
実行者 = CeleryExecutor
 sql_alchemy_conn = mysql+mysqldb://airflow:aaaa@$${MYSQL_HOST}:3306/airflow
 sql_engine_encoding = utf-8
 sql_alchemy_pool_enabled =有効 
 sql_alchemy_pool_size = 5
 sql_alchemy_max_overflow = 10
 sql_alchemy_pool_recycle = 1800
 sql_alchemy_pool_pre_ping =真 
 sql_alchemy_schema =
並列度 = 32
 1日あたりの最大アクティブタスク数 = 16
 dags_are_paused_at_creation =真 
 1日あたりの最大アクティブ実行数 = 16
 load_examples = True  
 load_default_connections = True  
プラグインフォルダ = /opt/airflow/plugins
 execute_tasks_new_python_interpreter = False  
 fernet_key =
 donot_pickle =真 
 dagbag_import_timeout = 30.0
 dagbag_import_error_tracebacks = True  
 dagbag_import_error_traceback_depth = 2
 dag_file_processor_timeout = 50
 task_runner = 標準タスクランナー
デフォルト偽装 =
セキュリティ =
ユニットテストモード = False  
 enable_xcom_pickle = False です 
強制終了したタスクのクリーンアップ時間 = 60
 dag_run_conf_overrides_params = True  
 dag_discovery_safe_mode =真 
デフォルトタスク再試行 = 0
 default_task_weight_rule = ダウンストリーム
最小シリアル化DAG更新間隔 = 30
最小シリアル化DAGフェッチ間隔 = 10
タスクあたりのレンダリングされた ti フィールドの最大数 = 30
 check_slas =真 
 xcom_backend = airflow.models.xcom.BaseXCom
 lazy_load_plugins = True  
 lazy_discover_providers = True  
最大データベース再試行回数 = 3
機密性の高いvar_conn_fieldsを非表示にする = True  
機密変数接続名 =
デフォルト_プール_タスク_スロット数 = 128
 [ログ]
ベースログフォルダ = /opt/airflow/logs
リモートログ = False  
リモートログ接続ID =
 google_key_path =
リモートベースログフォルダー =
 encrypt_s3_logs = False  
ログレベル = INFO
 fab_logging_level = 警告
ログ記録構成クラス =
色付きコンソールログ = True  
 coloured_log_format = [%%(blue)s%%(asctime)s%%(reset)s] {%%(blue)s%%(filename)s:%%(reset)s%%(lineno)d} %%(log_color)s%%(levelname)s%%(reset)s - %%(log_color)s%%(message)s%%(reset)s
色付きフォーマッタクラス = airflow.utils.log.colored_log.CustomTTYColoredFormatter
 log_format = [%%(asctime)s] {%%(ファイル名)s:%%(行番号)d} %%(レベル名)s - %%(メッセージ)s
 simple_log_format = %%(asctime)s %%(levelname)s - %%(message)s
タスクログプレフィックステンプレート =
 log_filename_template = {{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log
 log_processor_filename_template = {{ ファイル名 }}.log
 dag_processor_manager_log_location = /opt/airflow/logs/dag_processor_manager/dag_processor_manager.log
 task_log_reader = タスク
追加のロガー名 =
ワーカーログサーバーポート = 8793
 [メトリクス]
 statsd_on =偽 
 statsd_host = ローカルホスト
統計d_ポート = 8125
 statsd_prefix = エアフロー
統計d_allow_list =
統計名ハンドラ =
 statsd_datadog_enabled = False  
統計データドッグタグ =
 [秘密]
バックエンド =
バックエンド_kwargs =
 [クリ]
 api_client = airflow.api.client.local_client
エンドポイントURL = http://localhost:8080
 [デバッグ]
 fail_fast = False  
 [API]
 enable_experimental_api = False  
 auth_backend = airflow.api.auth.backend.deny_all
最大ページ制限 = 100
フォールバックページ制限 = 100
 google_oauth2_オーディエンス =
 google_key_path =
アクセス制御許可ヘッダー =
アクセス制御許可メソッド =
アクセス制御許可の起源 =
 [系統]
バックエンド =
 [アトラス]
 sasl_enabled = False  
ホスト =
ポート = 21000
ユーザー名 =
パスワード=
 [オペレーター]
 default_owner = エアフロー
デフォルトCPU = 1
デフォルト_ram = 512
デフォルトディスク = 512
デフォルト_gpus = 0
 default_queue =デフォルト 
 allow_illegal_arguments = False  
 [ハイブ]
デフォルト_hive_mapred_queue =
 [ウェブサーバー]
 base_url = https://devopsman.cn/airflow #Airflowドメイン名をカスタマイズ
default_ui_timezone = Asia/Shanghai # デフォルトのタイムゾーンを設定する
ウェブサーバーホスト = 0.0.0.0
ウェブサーバーポート = 8080
 web_server_ssl_cert =
ウェブサーバーのSSLキー =
ウェブサーバーマスタータイムアウト = 120
ウェブサーバーワーカータイムアウト = 120
ワーカー_リフレッシュ_バッチ_サイズ = 1
ワーカー更新間隔 = 6000
 reload_on_plugin_change = False  
シークレットキー = emEfndkf3QWZ5zVLE1kVMg==
労働者 = 4
ワーカークラス = 同期
アクセスログファイル = -
エラーログファイル = -
アクセスログフォーマット =
 expose_config = False  
公開ホスト名 = True  
 expose_stacktrace = True  
 dag_default_view = ツリー
dag_orientation = LR
ログフェッチタイムアウト秒 = 5
ログフェッチ遅延秒 = 2
ログ自動テーリングオフセット = 30
ログアニメーション速度 = 1000
デフォルトでは非表示の一時停止ダグス = False  
ページサイズ = 100
 navbar_color = #fff
デフォルトの dag 実行表示数 = 25
 enable_proxy_fix = False  
プロキシ_fix_x_for = 1
プロキシ_fix_x_proto = 1
プロキシ_fix_x_host = 1
プロキシ_fix_x_port = 1
プロキシ_fix_x_プレフィックス = 1
クッキーセキュア = False  
 cookie_samesite = 緩い
default_wrap = False  
 x_frame_enabled =有効 
 show_recent_stats_for_completed_runs = True  
 update_fab_perms =真 
セッションの存続時間 = 43200
自動更新間隔 = 3
 [メールアドレス]
 email_backend = airflow.utils.email.send_email_smtp
 email_conn_id = SMTP_デフォルト
デフォルトのメール再試行 = True  
デフォルトのメール失敗時 = True  
 [smtp] # メール設定
smtp_host = ローカルホスト
smtp_starttls =真 
 smtp_ssl =偽 
 SMTP_ポート = 25
 smtp_mail_from = [email protected]
 SMTP_タイムアウト = 30
 SMTP_再試行制限 = 5
 [歩哨]
監視オン =偽 
セントリー_dsn =
 [セロリ_kubernetes_executor]
 kubernetes_queue = kubernetes
 [セロリ]
 celery_app_name = airflow.executors.celery_executor
ワーカー同時実行数 = 16
ワーカー_umask = 0o077
 broker_url = redis://:xxxx@$${REDIS_HOST}:7480/0
 result_backend = db+mysql://airflow:aaaa@$${MYSQL_HOST}:3306/airflow
フラワーホスト = 0.0.0.0
花のURLプレフィックス =
フラワーポート = 5555
フラワー基本認証 =
同期並列性 = 0
 celery_config_options = airflow.config_templates.default_celery.DEFAULT_CELERY_CONFIG
 ssl_active =偽 
 SSLキー =
 ssl_cert =
 ssl_cacert =
プール = プリフォーク
オペレーションタイムアウト = 1.0
タスク追跡開始 = True  
タスク採用タイムアウト = 600
タスク公開の最大再試行回数 = 3
ワーカー事前チェック = False  
 [セロリブローカートランスポートオプション]
 [ダスク]
クラスターアドレス = 127.0.0.1:8786
 tls_ca =
 tls_cert =
 TLS_キー =
 [スケジューラ]
ジョブハートビート秒 = 5
スケジューラハートビート秒 = 5
実行回数 = -1
スケジューラ_アイドル_スリープ時間 = 1
最小ファイル処理間隔 = 30
 dag_dir_list_interval = 300
印刷統計間隔 = 30
プールメトリック間隔 = 5.0
スケジューラヘルスチェックしきい値 = 30
孤立したタスクのチェック間隔 = 300.0
 child_process_log_directory = /opt/airflow/logs/scheduler
スケジューラ_ゾンビ_タスク_しきい値 = 300
 catchup_by_default = True  
クエリあたりの最大数 = 512
行レベルロックの使用 = True  
ループごとに作成する最大ダグルン数 = 10
スケジュールごとのループの最大ダグル数 = 20
タスク実行後のスケジュール = True  
解析プロセス = 2
ファイル解析ソートモード = 変更日時
use_job_schedule = True  
 allow_trigger_in_future = False  
依存関係検出 = airflow.serialization.serialized_objects.DependencyDetector
トリガータイムアウトチェック間隔 = 15
 [トリガー]
デフォルト容量 = 1000
 [ケルベロス]
 ccache = /tmp/airflow_krb5_ccache
主 = 空気の流れ
再初期化頻度 = 3600
 kinit_path = kinit
キータブ = airflow.keytab
転送可能 = True  
 include_ip =真 
 [github_エンタープライズ]
 API_rev = v3 です
[エラスティックサーチ]
ホスト =
 log_id_template = {dag_id}-{task_id}-{execution_date}-{try_number}
 end_of_log_mark = ログ終了
フロントエンド =
 write_stdout = False  
 json_format = False  
 json_fields = asctime、ファイル名、行番号、レベル名、メッセージ
host_field = ホスト
offset_field = オフセット
[elasticsearch_configs]
 use_ssl = False  
 verify_certs = True  
 [Kubernetes]
ポッドテンプレートファイル =
ワーカーコンテナリポジトリ =
ワーカーコンテナタグ =
名前空間 =デフォルト 
ワーカーポッドを削除する = True  
失敗時にワーカーポッドを削除する = False  
ワーカーポッド作成バッチサイズ = 1
マルチネームスペースモード = False  
 in_cluster = True  
 kube_client_request_args=
削除オプション_kwargs =
 enable_tcp_keepalive =有効 
 tcp_keep_idle = 120
 tcp_keep_intvl = 30
 tcp_keep_cnt = 6
 verify_ssl =真 
ワーカーポッドの保留タイムアウト = 300
ワーカーポッドの保留タイムアウトチェック間隔 = 120
ワーカーポッドのキューチェック間隔 = 60
ワーカーポッドの保留タイムアウトバッチサイズ = 100
 [スマートセンサー]
スマートセンサーの使用 = False  
シャードコード上限 = 10000
破片 = 5
 sensors_enabled = 名前付きHiveパーティションセンサー

変更が完了したら、サービスを再起動します。

 docker-compose の再起動

4. データ同期

Airflow は 3 つのワーカーノードを使用するため、各ノードが構成を変更すると、他のノードが同期する必要があります。同時に、DAGS ディレクトリとプラグイン ディレクトリもリアルタイムで同期する必要があります。スケジューラが情報をノードにスケジュールした後、対応する DAGS ファイルが見つからない場合は、エラーが報告されます。したがって、リアルタイムのデータ同期には lsyncd を使用します。

 apt-get install lsyncd -y

公開鍵経由で接続するようにノードを構成する

ssh-keygen -t rsa -C "airflow-sync" -b 4096 #airflow-syncという名前のキーのペアを生成します
ipが100 200の場合、ssh-copy-id -i ~/.ssh/airflow-sync.pub ${USERNAME}@192.168.0.$ip -P12022 を実行します。完了

その後、秘密鍵を介して他のノードにアクセスできるようになります。

同期設定ファイルを編集し、lsyncd設定パラメータについて詳しく知るには、公式ドキュメント[2]に直接アクセスしてください。

設定 {
    logfile = "/var/log/lsyncd.log"、#ログファイル
    statusFile = "/var/log/lsyncd.status" 、# 同期ステータス情報
    pidfile = "/var/run/lsyncd.pid" 、
    ステータス間隔 = 1、
    nodaemon = false 、# デーモンプロセス
    inotifyMode = "CloseWrite" 、
    最大プロセス数 = 1、
    最大遅延 = 1、
 }
同期{
デフォルトの.rsync、
    ソース = "/data/airflow" 、
    ターゲット = "192.168.0.100:/data/airflow" 、 
 
    rsync = {
バイナリ= "/usr/bin/rsync" 、
       圧縮 = false 、
       アーカイブ = true 、
       所有者 = true 、
       権限 = true 、
 --delete = 真、  
       whole_file = false 、
       rsh = "/usr/bin/ssh -p 12022 -l suoper -o StrictHostKeyChecking=no -i /home/username/.ssh/airflow-rsync"  
    },
 }
同期{
デフォルトの.rsync、
    ソース = "/data/airflow" 、
    ターゲット = "192.168.0.200:/data/airflow" 、 
 
    rsync = {
バイナリ= "/usr/bin/rsync" 、
       圧縮 = false 、
       アーカイブ = true 、
       所有者 = true 、
       権限 = true 、
 --delete = 真、  
       whole_file = false 、
       rsh = "/usr/bin/ssh -p 12022 -l suoper -o StrictHostKeyChecking=no -i /home/username/.ssh/airflow-rsync"  
    },
 }

上記のパラメータの意味については、公式 Web サイトをご覧ください。ここでは、rsync の rsh を通じて ssh コマンドが定義されており、秘密鍵やカスタム ポートなどのセキュリティ対策が使用されるシナリオを解決できます。もちろん、パスワードなしのアクセスを構成して、構成に default.rsync または default.rsyncssh を使用することもできます。

lsyncd サービスホスティングを構成する

cat << EOF > /etc/systemd/system/lsyncd.service
 [ユニット]
説明=lsyncd
条件ファイルは実行可能=/usr/bin/lsyncd 
 
 =network-online.targetの後
欲しいもの=ネットワークオンライン.ターゲット
 
 [サービス]
開始リミットバースト=10
 ExecStart=/usr/bin/lsyncd /etc/lsyncd.conf
再起動=失敗時
再起動秒数=120
環境ファイル=-/etc/sysconfig/aliyun
キルモード=プロセス
[インストール]
 WantedBy =マルチユーザー.ターゲット
終了
 
 systemctlデーモンリロード
systemctl enable --now lsyncd.service #サービスを開始し、自動的に開始するように設定します 

これにより、データ (dag、プラグイン、airflow.cfg) の同期が完了します。後で CICD シナリオを使用する場合は、DAG ファイルを Bigdata1 ノードに直接アップロードすると、他の 2 つのノードが自動的に同期されます。問題が発生した場合は、ログを表示してデバッグすることができます。

 lsyncd -logすべての/etc/lsyncd.conf
末尾 -f /var/log/lsyncd.log

5リバースプロキシ[3]

https://lab.mycompany.com/myorg/airflow/ などのリバース プロキシの背後に Airflow を配置する必要がある場合は、次の構成でこれを行うことができます。

airflow.cfgでbase_urlを設定する

ベースURL = http://my_host/myorg/airflow
 enable_proxy_fix =有効 

nginx の設定

サーバー{
  聞く 80;
  サーバー名 lab.mycompany.com; 
 
  場所 /myorg/airflow/ {
      proxy_pass http://localhost:8080;
      proxy_set_header ホスト $http_host;
      proxy_redirectオフ;
      プロキシ_http_バージョン 1.1;
      proxy_set_header アップグレード $http_upgrade;
      proxy_set_header接続  「アップグレード」 ;
  }
 }

この時点で、Airflow 分散スケジューリング クラスターのインストールは基本的に完了です。具体的な効果は以下をご覧ください。

これをご覧になっているということは、あなたも Airflow を使用しているか、または Airflow に興味を持っているということです。ちなみに、Airflow の学習教材をいくつかお渡しします。

https://livebook.manning.com/book/data-pipelines-with-apache-airflow/chapter-12/1

参考文献

[1]Airflow 2.2.3 + MySQL8.0.27: https://mp.weixin.qq.com/s/VncpyXcTtlvnDkFrsAZ5lQ

[2]lsyncd設定ファイル: https://lsyncd.github.io/lsyncd/manual/config/file/

[3] プロキシの背後にあるエアフロー: https://airflow.apache.org/docs/apache-airflow/stable/howto/run-behind-proxy.html