Я пытаюсь преобразовать задание искровой отправки, которое настраивалось через файл yaml, в dag воздушного потока и запустить его через воздушный поток. Задание не выполняется, когда я пытаюсь смонтировать том и загрузить конфигурационную карту. Блок кода volumne mount выглядит примерно так в yaml
volumeMounts: — mountPath: /etc/jtp name: config-vol
И конфигурационная карта, которую нужно загрузить, выглядит примерно так
volumes: — configMap: name: qa-jtp-llt name: config-vol
Теперь, когда я пытаюсь перевести то же самое на свой даг, он превращается во что-то вроде этого
from kubernetes.client import models as k8s from airflow import DAG from datetime import timedelta from datetime import datetime from airflow.utils.dates import days_ago from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator from airflow.utils.trigger_rule import TriggerRule volume_mount = k8s.V1VolumeMount(name=’config-vol’, mount_path=’/etc/jtp’, sub_path=None, read_only=True) configmaps = [k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name=’config_name’))] volume = k8s.V1Volume( name=’config-vol’, persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name=’config-vol’), ) default_args = { ‘owner’: ‘airflow’, ‘depends_on_past’: False, ‘wait_for_downstream’: False, ‘start_date’: datetime(2020, 10, 8), ’email’: [‘[email protected]’], ’email_on_failure’: False, ’email_on_retry’: False, ‘max_active_runs’: 1, ‘catchup’: True } a dag = DAG( ‘sample_job’, default_args=default_args, description=’sample job to run on kubernetes-pod-operator’, schedule_interval=None, ) sample_job = KubernetesPodOperator( task_id=’task-1′, name=’sample-job’, namespace=’namespace1′, image=’image_name’, image_pull_policy=’Always’, volumes=[volume], volume_mounts=[volume_mount], configmaps=configmaps, cmds=[«/opt/entrypoint.sh», «driver»], arguments=[ ‘—class’, ‘something.thisthing.mainclass’, ‘myjar_file_name.jar’, ‘property_file_1.properties’, ‘property_file_2.properties’, ], dag=dag, )
Я пробую это на версии airflow 2.0.0 и использую оператор kubernetes pod. Проблема, с которой я сталкиваюсь, заключается в том, что когда я запускаю это задание в Airflow, это, в свою очередь, запускает локальный модуль для задания, который, в свою очередь, должен запускать другой модуль в пространстве имен, выполняющем задачу. Но второй модуль, который должен запустить задачу, не запускается, и сообщения об ошибке нет. Когда я удаляю все компоненты спецификации тома, такие как volume, volume_mount, config_map, появляется второй модуль в пространстве имен и выдает сообщение об ошибке о невозможности найти файл свойств. Что мне не хватает? Что я здесь делаю не так ?? Пробовали разные способы запустить это безуспешно, поэтому обратились за помощью к этому сообществу.
Спасибо
Источник: