如何使用 docker-compose 在分布式气流架构上配置 celery Worker？

Question

I\xe2\x80\x99m 设置一个分布式 Airflow 集群，其中除了 celery 工作线程之外的所有其他内容都在一台主机上运行，​​并且处理在多台主机上完成。Airflow2.0 设置是使用 Airflow 文档https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml中给出的 yaml 文件进行配置的。在我最初的测试中，当我在同一主机上运行所有内容时，架构可以很好地工作。问题是，如何在远程主机上启动 celery 工作程序？

\n

到目前为止，我尝试创建上述 docker-compose 的精简版本，其中我只在工作主机上启动 celery 工作人员，而没有其他任何东西。但我遇到了数据库连接的一些问题。在修剪版本中，我更改了 URL，以便它们指向运行 db 和 redis 的主机。

\n

dags、日志、插件和 postgresql 数据库位于所有主机都可见的共享驱动器上。

\n

我应该如何进行配置？有什么想法要检查什么吗？连接等？\nCelery 工作人员 docker-compose 配置：

\n

---\nversion: \'3\'\nx-airflow-common:\n  &airflow-common\n  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}\n  environment:\n    &airflow-common-env\n    AIRFLOW_UID: 50000\n    AIRFLOW_GID: 50000\n    AIRFLOW__CORE__EXECUTOR: CeleryExecutor\n    AIRFLOW__CORE__SQL_ALCHEMY_CONN: \npostgresql+psycopg2://airflow:airflow@airflowhost.example.com:8080/airflow\n    AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow@airflowhost.example.com:8080/airflow\n    AIRFLOW__CELERY__BROKER_URL: redis://:@airflow@airflowhost.example.com:6380/0\n    AIRFLOW__CORE__FERNET_KEY: \'\'\n    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: \'true\'\n    AIRFLOW__CORE__LOAD_EXAMPLES: \'true\'\n    AIRFLOW__API__AUTH_BACKEND: \'airflow.api.auth.backend.basic_auth\'\n    REDIS_PORT: 6380\n   volumes:\n    - /airflow/dev/dags:/opt/airflow/dags\n    - /airflow/dev/logs:/opt/airflow/logs\n    - /airflow/dev/plugins:/opt/airflow/plugins\n   user: "${AIRFLOW_UID:-50000}:${AIRFLOW_GID:-50000}"\nservices:\n  airflow-remote-worker:\n    <<: *airflow-common\n    command: celery worker\n    healthcheck:\n      test:\n        - "CMD-SHELL"\n        - \'celery --app airflow.executors.celery_executor.app inspect ping -d "celery@$${HOSTNAME}"\'\n      interval: 10s\n      timeout: 10s\n      retries: 5\n    restart: always\n

\n

编辑 1:\n我对日志文件仍然遇到一些困难。看来共享日志目录并不能解决丢失日志文件的问题。我按照建议在 main 上添加了 extra_host 定义，并在工作计算机上打开了端口 8793。\n工作任务失败并显示以下日志：

\n

*** Log file does not exist: \n/opt/airflow/logs/tutorial/print_date/2021-07- \n01T13:57:11.087882+00:00/1.log\n*** Fetching from: http://:8793/log/tutorial/print_date/2021-07-01T13:57:11.087882+00:00/1.log\n*** Failed to fetch log file from worker. Unsupported URL protocol \'\'\n

\n

Answer 1

这些设置远不是“最终设置”，而是在核心节点和工作人员中使用 Airflow 的 docker-compose 对我有用的一些设置：

主节点：

• 工作节点必须可以从Webserver运行的主节点访问。我发现这张架构图对于CeleryExecutor理清问题非常有帮助。

  当尝试读取日志时，如果在本地找不到日志，它将尝试从远程工作人员检索日志。因此，您的主节点可能不知道工作人员的主机名，因此您要么更改主机名的解析方式（hostname_callable设置，默认为socket.getfqdn），要么只是简单地将名称解析功能添加到Webserver. extra_hosts这可以通过在定义中添加配置键来完成x-airflow-common：

---
version: "3"
x-airflow-common: &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}
  environment: &airflow-common-env
    ...# env vars
  extra_hosts:
    - "worker-01-hostname:worker-01-ip-address" # "worker-01-hostname:192.168.0.11"
    - "worker-02-hostname:worker-02-ip-address"

*请注意，在您有共享驱动器的特定情况下，因此我认为日志将在本地找到。

• 定义并行性、DAG 并发性和调度程序解析过程。可以通过使用环境变量来完成：

x-airflow-common: &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}
  environment: &airflow-common-env
    AIRFLOW__CORE__PARALLELISM: 64
    AIRFLOW__CORE__DAG_CONCURRENCY: 32
    AIRFLOW__SCHEDULER__PARSING_PROCESSES: 4

当然，要设置的值取决于您的具体情况和可用资源。这篇文章很好地概述了该主题。DAG 设置也可以在定义时被覆盖DAG。

工作节点：

• 定义worker CELERY__WORKER_CONCURRENCY，默认值可以是机器上可用的CPU数量（文档）。

• 定义如何访问主节点中运行的服务。设置 IP 或主机名并注意主节点中匹配的公开端口：

x-airflow-common: &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}
  environment: &airflow-common-env
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CELERY__WORKER_CONCURRENCY: 8
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@main_node_ip_or_hostname:5432/airflow # 5432 is default postgres port
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@main_node_ip_or_hostname:5432/airflow
  AIRFLOW__CELERY__BROKER_URL: redis://:@main_node_ip_or_hostname:6379/0

• 共享从“.env”文件中读取的相同Fernet 密钥和秘密密钥：

  environment: &airflow-common-env
    AIRFLOW__CORE__FERNET_KEY: ${FERNET_KEY}
    AIRFLOW__WEBSERVER__SECRET_KEY: ${SECRET_KEY}

  env_file:
    - .env

.env 文件：FERNET_KEY=jvYUaxxxxxxxxxxxxx=

• 集群中的每个节点（主节点和工作节点）都应用相同的设置，这一点至关重要。

• 为工作服务定义主机名，以避免自动生成与容器 ID 匹配的主机名。

• 公开端口 8793，这是用于从工作程序获取日志的默认端口（文档）：

services:
  airflow-worker:
    <<: *airflow-common
    hostname: ${HOSTNAME}
    ports:
      - 8793:8793
    command: celery worker
    restart: always

• 确保每个工作节点主机都以相同的时间配置运行，几分钟的差异可能会导致严重的执行错误，而这些错误可能不太容易发现。考虑在主机操作系统上启用 NTP 服务。

如果您的工作负载很重并且并发性很高，您可能需要调整 Postgres 设置，例如max_connections和shared_buffers。这同样适用于主机操作系统网络设置，例如ip_local_port_range或somaxconn。

在我在初始集群设置过程中遇到的任何问题中，Flower工作执行日志始终提供有用的详细信息和错误消息，包括任务级日志和 Docker-Compose 服务日志，即：docker-compose logs --tail=10000 airflow-worker > worker_logs.log。

希望这对你有用！