Блог

У нас работает воздушный поток (с помощью Docker compose) с несколькими активными DAG. На прошлой неделе мы обновили наш воздушный поток до версии 2.1.3. Это привело к ошибке для группы обеспечения доступности баз данных, в которой мы используем DockerOperator:

 airflow.exceptions.AirflowException: Invalid arguments were passed to DockerOperator (task_id: t_docker). Invalid arguments were:
**kwargs: {'volumes':
 

Я нашел эту заметку о выпуске, в которой говорится, что

Параметр объемы в файле airflow. providers.docker.operators.docker.DockerОператор и airflow.провайдеры.docker.операторы.docker_сварм.DockerSwarmOperator был заменен параметром mounts

Так что я поменял наш DAG с

 t_docker = DockerOperator(
        task_id='t_docker',
        image='customimage:latest',
        container_name='custom_1',
        api_version='auto',
        auto_remove=True,
        volumes=['/home/airflow/scripts:/opt/airflow/scripts','/home/airflow/data:/opt/airflow/data'],
        docker_url='unix://var/run/docker.sock',
        network_mode='bridge',
        dag=dag
    )
 

к этому

 t_docker = DockerOperator(
        task_id='t_docker',
        image='customimage:latest',
        container_name='custom_1',
        api_version='auto',
        auto_remove=True,
        mounts=['/home/airflow/scripts:/opt/airflow/scripts','/home/airflow/data:/opt/airflow/data'],
        docker_url='unix://var/run/docker.sock',
        network_mode='bridge',
        dag=dag
    )
 

Но теперь я получаю эту ошибку:

 docker.errors.APIError: 500 Server Error for http docker://localhost/v1.41/containers/create?name=custom_1: Internal Server Error ("json: cannot unmarshal string into Go struct field HostConfig.HostConfig.Mounts of type mount.Mount")
 

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

Изменение касается не только имени параметра, но и синтаксиса монтирования.

Вы должны заменить

 volumes=['/home/airflow/scripts:/opt/airflow/scripts','/home/airflow/data:/opt/airflow/data']
 

с:

 mounts=[
    Mount(source="/home/airflow/scripts", target="/opt/airflow/scripts", type="bind"),
    Mount(source="/home/airflow/data", target="/opt/airflow/data", type="bind"),
]
 

Таким образом, ваш код будет:

 from docker.types import Mount
t_docker = DockerOperator(
    task_id='t_docker',
    image='customimage:latest',
    container_name='custom_1',
    api_version='auto',
    auto_remove=True,
    mounts=[
        Mount(source="/home/airflow/scripts", target="/opt/airflow/scripts", type="bind"),
        Mount(source="/home/airflow/data", target="/opt/airflow/data", type="bind"),
    ],
    docker_url='unix://var/run/docker.sock',
    network_mode='bridge',
    dag=dag
)