работает KubernetesPodOperator

Question

Я запускаю поток воздуха в Kubernetes и использую «Оператор Kubernetes» . Когда я запускаю BashOperator или PythonOperator , он работает отлично Использование:

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

Когда я пытаюсь запустить run dag с KubernetesPodOperator этотерпит неудачу

например:

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
)

Я вижу, что созданное изображение докера - это не изображение, которое я указал выше (ubuntu: 18.04), а изображение по умолчанию изконфигурация (AIRFLOW__KUBERNETES__WORKER_CONTAINER_REPOSITORY)

в журнале планировщика. Я вижу:

[2019-10-06 12: 59: 56,279] {{scheduler_job.py:921}} INFO - 1 задание готово к выполнению: [2019-10-06 12: 59: 56,325] {{scheduler_job.py:953}} INFO - определение задач для запуска в пуле (name = default_pool) со 128 открытыми слотамии 1 экземпляр задачи готов к постановке в очередь [2019-10-06 12: 59: 56,326] {{scheduler_job.py:981}} INFO - у оператора DAG есть 0/16 запущенных и поставленных в очередь задач [2019-10-06 12:59: 56,361] {{scheduler_job.py:1031}} INFO - перевод следующих задач в состояние очереди: [2019-10-06 12: 59: 56,398] {{scheduler_job.py:1107}} INFO - Установка следующих 1 задач в состояние очереди: [2019-10-06 12: 59: 56,401] {{scheduler_job.py:1143}} INFO - Отправка ('koperator', 'task', datetime.datetime (2019, 10, 6, 12, 59, 50, 146375, tzinfo =), 1) исполнителю с приоритетом 1 и очередью по умолчанию [2019-10-06 12: 59: 56,403] {{base_executor.py: 59}} INFO - Добавление в очередь: ['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}}ИНФОРМАЦИЯ - Добавьте задачу ('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 '] с executor_config {} [2019-10-06 12: 59: 56,417] {{kubernetes_executor.py:441}} INFO - задание Kubernetes ((' koperator ',' task ', datetime. даtetime (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 = Нет, image_pull_policy)= Нет, request_memory = Нет, request_cpu = Нет, limit_memory = Нет, limit_cpu = Нет, limit_gpu = Нет, gcp_service_account_key = Нет, node_selectors = Нет, сходство = Нет, аннотации = {}, тома = [], volume_mountsдопуски = нет, метки = {})) [2019-10-06 12: 59: 56,498] {{kubernetes_executor.py:353}} INFO - событие: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d имело событие типа ДОБАВЛЕНО [2019-10-0612: 59: 56,509] {{kubernetes_executor.py:385}} ИНФОРМАЦИЯ - Событие: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d Ожидается [2019-10-06 12: 59: 56,528] {{kubernetes_executor.py:tk} Оператор INFO: операционная система} INFO2f35f3b347a149bcb2133ef58cf9e77d было событие типа MODIFIED [2019-10-06 12: 59: 56,529] {{kubernetes_executor.py:385}} ИНФО- Событие: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d В ожидании [2019-10-06 12: 59: 56543] {{kubernetes_executor.py:353}} ИНФО - Событие: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d было событие типа МОДИФИЦИРОВАННЫХ [2019-10-06 12:59: 56 544событие типа MODIFIED [2019-10-06 12: 59: 59,492] {{kubernetes_executor.py:393}} INFO - событие: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d выполняется [2019-10-06 13: 00: 10,873] {{kuber. py: 353}} INFO - Событие: у koperatortask-2f35f3b347a149bcb2133ef58cf9e77d было событие типа MODIFIED [2019-10-06 13: 00: 10,874] {{kubernetes_executor.py:390}} ИНФОРМАЦИЯ - Событие: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d Успешно [2019-10-06 13: 00: 12,236] {{kubernetes_exe34949 ::- Попытка доделать стручок;pod_id: koperatortask-2f35f3b347a149bcb2133ef58cf9e77d;состояние: нет;метки: {'airflow-worker': 'b46fd37e-959c-4844-81e1-dff9df2e98e2', 'dag_id': 'koperator', 'execute_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 - Проверка 1 экземпляров задачи. [2019-10-06 13: 00: 12,247] {{kubernetes_executor.py:626}} ИНФОРМАЦИЯ - найдено совпадающее задание koperator-task (2019-10-06 12: 59: 50.146375 + 00: 00) с текущим состоянием up_for_retry[2019-10-06 13: 00: 12,253] {{kubernetes_executor.py:783}} INFO - Изменение состояния (('оператор', 'задача', datetime.datetime (2019, 10, 6, 12, 59,50, 146375, tzinfo = tzlocal ()), 1), None, 'koperatortask-2f35f3b347a149bcb2133ef58cf9e77d', '34894988') до None [2019-10-06 13: 00: 12,273] {{scheduler_job.py:12}}- Исполнитель сообщает о выполнении koperator.task execute_date = 2019-10-06 12: 59: 50.146375 + 00: 00 с состоянием None для

журнал поднятого модуля:

[2019-10-06 12: 02: 11,961] {{ init .py: 51}} INFO - Использование executor LocalExecutor [2019-10-06 12:02: 12,844] {{dagbag.py:90}} ИНФОРМАЦИЯ - Заполнение DagBag из /usr/local/airflow/dags/KubernetesPodOperator.py [2019-10-06 12: 02: 13,571] {{cli.py:516}} ИНФОРМАЦИЯ - Запуск на хосте koperatortask-bd0c81d6039c4b329ae8dd2292c0c566

что я делаю не так?

как мне запустить dag на kubernetes с помощью KubernetesPodOperator?

спасибо, Aviad