kubernetes - 运行 KubernetesPodOperator

Question

我在 Kubernetes 上运行 Airflow 并使用 “Kubernetes 运营商” .
当我运行 BashOperator 或 Python运算符(operator)它有效很好
使用:

executor_config = {
    "KubernetesExecutor": {
    "image": "image_with_airflow_and_my_code:latest"        
    }
}

当我尝试使用 KubernetesPodOperator 运行 run dag 时，它会失败

例如:

k = KubernetesPodOperator(namespace='default',
    image="ubuntu:18.04",
    cmds=["bash", "-cx"],
    arguments=["echo", "10"],
    name="test",
    task_id="task",
    is_delete_operator_pod=False,
    dag=dag
)

我看到创建的 docker 镜像不是我上面指定的镜像(ubuntu:18.04)，而是配置中的默认镜像(AIRFLOW__KUBERNETES__WORKER_CONTAINER_REPOSITORY)

在调度程序日志中我看到:

[2019-10-06 12:59:56,279] {{scheduler_job.py:921}} INFO - 1 tasks up for execution: [2019-10-06 12:59:56,325] {{scheduler_job.py:953}} INFO - Figuring out tasks to run in Pool(name=default_pool) with 128 open slots and 1 task instances ready to be queued [2019-10-06 12:59:56,326] {{scheduler_job.py:981}} INFO - DAG koperator has 0/16 running and queued tasks [2019-10-06 12:59:56,361] {{scheduler_job.py:1031}} INFO - Setting the following tasks to queued state: [2019-10-06 12:59:56,398] {{scheduler_job.py:1107}} INFO - Setting the following 1 tasks to queued state: [2019-10-06 12:59:56,401] {{scheduler_job.py:1143}} INFO - Sending ('koperator', 'task', datetime.datetime(2019, 10, 6, 12, 59, 50, 146375, tzinfo=), 1) to executor with priority 1 and queue default [2019-10-06 12:59:56,403] {{base_executor.py:59}} INFO - Adding to queue: ['airflow', 'run', 'koperator', 'task', '2019-10-06T12:59:50.146375+00:00', '--local', '--pool', 'default_pool', '-sd', '/usr/local/airflow/dags/KubernetesPodOperator.py'] [2019-10-06 12:59:56,405] {{kubernetes_executor.py:764}} INFO - Add task ('koperator', 'task', datetime.datetime(2019, 10, 6, 12, 59, 50, 146375, tzinfo=), 1) with command ['airflow', 'run', 'koperator', 'task', '2019-10-06T12:59:50.146375+00:00', '--local', '--pool', 'default_pool', '-sd', '/usr/local/airflow/dags/KubernetesPodOperator.py'] with executor_config {} [2019-10-06 12:59:56,417] {{kubernetes_executor.py:441}} INFO - Kubernetes job is (('koperator', 'task', datetime.datetime(2019, 10, 6, 12, 59, 50, 146375, tzinfo=), 1), ['airflow', 'run', 'koperator', 'task', '2019-10-06T12:59:50.146375+00:00', '--local', '--pool', 'default_pool', '-sd', '/usr/local/airflow/dags/KubernetesPodOperator.py'], KubernetesExecutorConfig(image=None, image_pull_policy=None, request_memory=None, request_cpu=None, limit_memory=None, limit_cpu=None, limit_gpu=None, gcp_service_account_key=None, node_selectors=None, affinity=None, annotations={}, volumes=[], volume_mounts=[], tolerations=None, labels={})) [2019-10-06 12:59:56,498] {{kubernetes_executor.py:353}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d had an event of type ADDED [2019-10-06 12:59:56,509] {{kubernetes_executor.py:385}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d Pending [2019-10-06 12:59:56,528] {{kubernetes_executor.py:353}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d had an event of type MODIFIED [2019-10-06 12:59:56,529] {{kubernetes_executor.py:385}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d Pending [2019-10-06 12:59:56,543] {{kubernetes_executor.py:353}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d had an event of type MODIFIED [2019-10-06 12:59:56,544] {{kubernetes_executor.py:385}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d Pending [2019-10-06 12:59:59,492] {{kubernetes_executor.py:353}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d had an event of type MODIFIED [2019-10-06 12:59:59,492] {{kubernetes_executor.py:393}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d is Running [2019-10-06 13:00:10,873] {{kubernetes_executor.py:353}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d had an event of type MODIFIED [2019-10-06 13:00:10,874] {{kubernetes_executor.py:390}} INFO - Event: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d Succeeded [2019-10-06 13:00:12,236] {{kubernetes_executor.py:493}} INFO - Attempting to finish pod; pod_id: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d; state: None; labels: {'airflow-worker': 'b46fd37e-959c-4844-81e1-dff9df2e98e2', 'dag_id': 'koperator', 'execution_date': '2019-10-06T12_59_50.146375_plus_00_00', 'task_id': 'task', 'try_number': '1'} [2019-10-06 13:00:12,245] {{kubernetes_executor.py:616}} INFO - Checking 1 task instances. [2019-10-06 13:00:12,247] {{kubernetes_executor.py:626}} INFO - Found matching task koperator-task (2019-10-06 12:59:50.146375+00:00) with current state of up_for_retry [2019-10-06 13:00:12,253] {{kubernetes_executor.py:783}} INFO - Changing state of (('koperator', 'task', datetime.datetime(2019, 10, 6, 12, 59, 50, 146375, tzinfo=tzlocal()), 1), None, 'koperatortask-2f35f3b347a149bcb2133ef58cf9e77d', '34894988') to None [2019-10-06 13:00:12,273] {{scheduler_job.py:1283}} INFO - Executor reports execution of koperator.task execution_date=2019-10-06 12:59:50.146375+00:00 exited with status None for

pods 日志:

[2019-10-06 12:02:11,961] {{init.py:51}} INFO - Using executor LocalExecutor [2019-10-06 12:02:12,844] {{dagbag.py:90}} INFO - Filling up the DagBag from /usr/local/airflow/dags/KubernetesPodOperator.py [2019-10-06 12:02:13,571] {{cli.py:516}} INFO - Running on host koperatortask-bd0c81d6039c4b329ae8dd2292c0c566

我究竟做错了什么？

如何使用 KubernetesPodOperator 在 kubernetes 上运行 dag？

谢谢， 阿维亚德

Answer 1

不幸的是，我没有看到足够的信息来确定问题所在。
添加参数。

“get_logs”:真

到 KubernetesPodOperator。

这样，运行会将来自 KubernetesExecutor 和 KubernetesPodOperator pod 的标准输出合并到 Airflow 任务日志中。

它应该让您更清楚地了解正在发生的事情。