气流在 Docker 中运行其他容器的作业答案

【问题标题】：Airflow Running Other Container's Job in Docker气流在 Docker 中运行其他容器的作业
【发布时间】：2022-08-16 19:09:49
【问题描述】：

我尝试在 Airflow 中使用 PythonOperator 运行 (python_callable) 以执行其他容器的作业 (PHP)。但如果失败，主要错误在这里：

Broken DAG: [/opt/airflow/dags/cobalagi.py] Traceback (most recent call last):
  File \"<frozen importlib._bootstrap>\", line 219, in _call_with_frames_removed
  File \"/opt/airflow/dags/cobalagi.py\", line 3, in <module>
    import command
ModuleNotFoundError: No module named \'command\'

尝试删除“导入命令”，但气流图变为红色（失败），错误日志如下：

[2022-08-16, 04:31:52 UTC] {standard_task_runner.py:97} ERROR - Failed to execute job 1055 for task address (Object of type bytes is not JSON serializable; 15679)

权限被拒绝错误。尝试重新启动桌面，但仍然：

[2022-08-16, 11:05:40 UTC] {standard_task_runner.py:97} ERROR - Failed to execute job 2046 for task address (Error while fetching server API version: (\'Connection aborted.\', PermissionError(13, \'Permission denied\')); 740)

然后我在 docker-compose.yml 的目录中用 \"pip install command\" 安装 \"command\" 但错误是一样的。这是源代码：

version: \'2.3.2\'
x-airflow-common:
  &airflow-common
  # In order to add custom dependencies or upgrade provider packages you can use your extended image.
  # Comment the image line, place your Dockerfile in the directory where you placed the docker-compose.yaml
  # and uncomment the \"build\" line below, Then run `docker-compose build` to build the images.
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.3.2}
  # build: .
  environment:
    &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: LocalExecutor
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    # For backward compatibility, with Airflow <2.3
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CORE__FERNET_KEY: \'\'
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: \'true\'
    AIRFLOW__CORE__LOAD_EXAMPLES: \'false\'
    AIRFLOW__API__AUTH_BACKENDS: \'airflow.api.auth.backend.basic_auth\'
    _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:- airflow-code-editor apache-airflow-providers-docker}
  volumes:
    - ./dags:/opt/airflow/dags
    - ./logs:/opt/airflow/logs
    - ./plugins:/opt/airflow/plugins
    - //var/run/docker.sock:/var/run/docker.sock
  user: \"${AIRFLOW_UID:-50000}:0\"
  depends_on:
    &airflow-common-depends-on
    postgres:
      condition: service_healthy

services:
  postgres:
    image: postgres:13
    ports:
      - 5432:5432
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    volumes:
      - postgres-db-volume:/var/lib/postgresql/data
      - //var/run/docker.sock:/var/run/docker.sock
    healthcheck:
      test: [\"CMD\", \"pg_isready\", \"-U\", \"airflow\"]
      interval: 5s
      retries: 5
    restart: always

  airflow-webserver:
    <<: *airflow-common
    command: webserver
    ports:
      - 8080:8080
    healthcheck:
      test: [\"CMD\", \"curl\", \"--fail\", \"http://localhost:8080/health\"]
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-scheduler:
    <<: *airflow-common
    command: scheduler
    healthcheck:
      test: [\"CMD-SHELL\", \'airflow jobs check --job-type SchedulerJob --hostname \"$${HOSTNAME}\"\']
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-triggerer:
    <<: *airflow-common
    command: triggerer
    healthcheck:
      test: [\"CMD-SHELL\", \'airflow jobs check --job-type TriggererJob --hostname \"$${HOSTNAME}\"\']
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-init:
    <<: *airflow-common
    entrypoint: /bin/bash
    # yamllint disable rule:line-length
    command:
      - -c
      - |
        function ver() {
          printf \"%04d%04d%04d%04d\" $${1//./ }
        }
        airflow_version=$$(gosu airflow airflow version)
        airflow_version_comparable=$$(ver $${airflow_version})
        min_airflow_version=2.2.0
        min_airflow_version_comparable=$$(ver $${min_airflow_version})
        if (( airflow_version_comparable < min_airflow_version_comparable )); then
          echo
          echo -e \"\\033[1;31mERROR!!!: Too old Airflow version $${airflow_version}!\\e[0m\"
          echo \"The minimum Airflow version supported: $${min_airflow_version}. Only use this or higher!\"
          echo
          exit 1
        fi
        if [[ -z \"${AIRFLOW_UID}\" ]]; then
          echo
          echo -e \"\\033[1;33mWARNING!!!: AIRFLOW_UID not set!\\e[0m\"
          echo \"If you are on Linux, you SHOULD follow the instructions below to set \"
          echo \"AIRFLOW_UID environment variable, otherwise files will be owned by root.\"
          echo \"For other operating systems you can get rid of the warning with manually created .env file:\"
          echo \"    See: https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html#setting-the-right-airflow-user\"
          echo
        fi
        one_meg=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=\"false\"
        if (( mem_available < 4000 )) ; then
          echo
          echo -e \"\\033[1;33mWARNING!!!: Not enough memory available for Docker.\\e[0m\"
          echo \"At least 4GB of memory required. You have $$(numfmt --to iec $$((mem_available * one_meg)))\"
          echo
          warning_resources=\"true\"
        fi
        if (( cpus_available < 2 )); then
          echo
          echo -e \"\\033[1;33mWARNING!!!: Not enough CPUS available for Docker.\\e[0m\"
          echo \"At least 2 CPUs recommended. You have $${cpus_available}\"
          echo
          warning_resources=\"true\"
        fi
        if (( disk_available < one_meg * 10 )); then
          echo
          echo -e \"\\033[1;33mWARNING!!!: Not enough Disk space available for Docker.\\e[0m\"
          echo \"At least 10 GBs recommended. You have $$(numfmt --to iec $$((disk_available * 1024 )))\"
          echo
          warning_resources=\"true\"
        fi
        if [[ $${warning_resources} == \"true\" ]]; then
          echo
          echo -e \"\\033[1;33mWARNING!!!: You have not enough resources to run Airflow (see above)!\\e[0m\"
          echo \"Please follow the instructions to increase amount of resources available:\"
          echo \"   https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html#before-you-begin\"
          echo
        fi
        mkdir -p /sources/logs /sources/dags /sources/plugins
        chown -R \"${AIRFLOW_UID}:0\" /sources/{logs,dags,plugins}
        exec /entrypoint airflow version
    # yamllint enable rule:line-length
    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:-airflow}
      _PIP_ADDITIONAL_REQUIREMENTS: \'\'
    user: \"0:0\"
    volumes:
      - .:/sources
      - //var/run/docker.sock:/var/run/docker.sock

  airflow-cli:
    <<: *airflow-common
    profiles:
      - debug
    environment:
      <<: *airflow-common-env
      CONNECTION_CHECK_MAX_COUNT: \"0\"
    command:
      - bash
      - -c
      - airflow
volumes:
  postgres-db-volume:

这是 python.py ：

import docker

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.python_operator import PythonOperator


from command import showAddress

default_args = {
    \"owner\": \"airflow\", 
    \"start_date\": datetime(2021, 3, 7)}

def showAddress():
    client = docker.from_env()
    container = client.containers.get(\'shouts-laravel-app\')
    cmd = container.exec_run(
        \"php artisan showAddress\"
    )

    return cmd

with DAG(
    dag_id=\"tryToShow\", 
    default_args=default_args, 
    schedule_interval=\'@daily\'
    
) as dag:

    showAddress = PythonOperator(
        task_id=\"address\",
        python_callable=showAddress
    )

    showAddress

我已经在 1 个名为 laravel_laravel-shouts 的网络中制作了这些容器。这是码头工人：

DOCKER LIST

请帮助，我应该怎么做才能达到目标（使用气流运行 PHP 作业）？

标签： python php docker containers airflow

【解决方案1】：

在某些情况下，方法exec_run返回字节，当你在一个PythonOperator中调用方法showAddress时，这个方法的结果会存储在一个xcom中，所以它需要是可序列化的，但是字节是不是，所以 Airflow 会引发错误：

ERROR - Failed to execute job 1055 for task address (Object of type bytes is not JSON serializable; 15679)

尝试将元组cmd 中的数据转换为字符串：here 是一种方法。

# just replace
# return cmd
# by
return cmd[1].decode('utf-8') if isinstance(cmd[1], bytes) else cmd[1]

【讨论】：

谢谢你，先生。但我很困惑，因为基本上我是 PHP 学习者。我认为你给出的链接更像是简单的打印。就我而言，它从数据库/数组中调用数据
我刚刚通过添加您可以使用的代码更新了我的答案
谢谢你，先生。它的许可被拒绝。你有什么解决办法吗？在互联网上找到了一些，但它是关于 DockerOperator 的，我的是 PythonOperator
我不明白，现在你可以访问输出了，但是访问被拒绝了？
我不知道它的工作与否。错误更改为“权限被拒绝”。工作刚刚变成红色。我会更新上面的错误