Оператору документооборота Airflow не удалось найти файл .sock на локальном компьютере - PullRequest
Новая
       6

Оператору документооборота Airflow не удалось найти файл .sock на локальном компьютере

0 голосов
/ 31 октября 2018

Я хочу запустить Docker-контейнер, содержащий скрипт Python, по расписанию с Airflow. У меня возникла проблема при локальном запуске задачи DockerOperator через CLI Airflow. 

--------------------------------------------------------------------------------
Starting attempt 1 of 4
--------------------------------------------------------------------------------

[2018-10-31 15:20:10,760] {models.py:1569} INFO - Executing <Task(DockerOperator): amplitude_to_s3_docker> on 2018-10-02T00:00:00+00:00
[2018-10-31 15:20:10,761] {base_task_runner.py:124} INFO - Running: ['bash', '-c', 'airflow run get_amplitude_docker_dag amplitude_to_s3_docker 2018-10-02T00:00:00+00:00 --job_id 19 --raw -sd DAGS_FOLDER/amplitude_to_s3_docker_dag.py --cfg_path /var/folders/ys/83xq3b3d1qv3zfx3dtkkp9tc0000gn/T/tmp_lu9mgzz']
[2018-10-31 15:20:12,501] {base_task_runner.py:107} INFO - Job 19: Subtask amplitude_to_s3_docker [2018-10-31 15:20:12,501] {__init__.py:51} INFO - Using executor SequentialExecutor
[2018-10-31 15:20:13,465] {base_task_runner.py:107} INFO - Job 19: Subtask amplitude_to_s3_docker [2018-10-31 15:20:13,464] {models.py:258} INFO - Filling up the DagBag from /Users/thisuser/Projects/GitRepos/DataWarehouse/dags/amplitude_to_s3_docker_dag.py
[2018-10-31 15:20:13,581] {base_task_runner.py:107} INFO - Job 19: Subtask amplitude_to_s3_docker [2018-10-31 15:20:13,581] {example_kubernetes_operator.py:54} WARNING - Could not import KubernetesPodOperator: No module named 'kubernetes'
[2018-10-31 15:20:13,582] {base_task_runner.py:107} INFO - Job 19: Subtask amplitude_to_s3_docker [2018-10-31 15:20:13,582] {example_kubernetes_operator.py:55} WARNING - Install kubernetes dependencies with:     pip install airflow['kubernetes']
[2018-10-31 15:20:13,770] {base_task_runner.py:107} INFO - Job 19: Subtask amplitude_to_s3_docker [2018-10-31 15:20:13,770] {cli.py:492} INFO - Running <TaskInstance: get_amplitude_docker_dag.amplitude_to_s3_docker 2018-10-02T00:00:00+00:00 [running]> on host 254.1.168.192.in-addr.arpa
[2018-10-31 15:20:13,804] {docker_operator.py:169} INFO - Starting docker container from image amplitude
[2018-10-31 15:20:13,974] {models.py:1736} ERROR - create_container() got an unexpected keyword argument 'cpu_shares'
Traceback (most recent call last):
  File "/Users/thisuser/anaconda/lib/python3.5/site-packages/airflow/models.py", line 1633, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/Users/thisuser/anaconda/lib/python3.5/site-packages/airflow/operators/docker_operator.py", line 210, in execute
    working_dir=self.working_dir
TypeError: create_container() got an unexpected keyword argument 'cpu_shares'

У меня есть скрипт, работающий вне Airflow fine, используя эту команду:

docker run amplitude get_amplitude.py 2018-10-02 2018-10-02

Вот мой даг и файл задания: 

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.docker_operator import DockerOperator
from datetime import datetime, timedelta


default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2018, 10, 30),
    "email": ["me@myemail.com"],
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG("get_amplitude_docker_dag", default_args=default_args, schedule_interval=timedelta(minutes=10))

templated_command = """
    get_amplitude.py {{ ds }} {{ ds }}
"""

t1 = DockerOperator(
   task_id='amplitude_to_s3_docker',
   command=templated_command,
   image='amplitude',
   dag=dag
)

После инициализации локальной базы данных воздушного потока и запуска планировщика webserver + я запускаю свою задачу dag с помощью:

airflow run get_amplitude_docker_dag amplitude_to_s3_docker 2018-10-02

Кроме того, задача будет нормально выполняться через поток воздуха, если я настрою ее как оператор bash: 

templated_command = """
   docker run amplitude get_amplitude.py {{ ds }} {{ ds }} 
"""


t1 = BashOperator(
    task_id="amplitude_to_s3",
    bash_command=templated_command,
    params={},
    dag=dag,
)

Я читал ранее, что могут быть проблемы с монтированием демона docker, но мой .sock-файл находится там, куда указывает параметр docker_url по умолчанию, /var/run/docker.sock.

Может кто-нибудь помочь мне настроить эту работу?

1 Ответ

0 голосов
/ 19 ноября 2018

Фактическая ошибка TypeError: create_container () получил неожиданный аргумент ключевого слова 'cpu_shares' , что означает, что функция create_container не ожидает cpu_shares в качестве аргумента.

У меня такая же ошибка при использовании библиотеки докера Python версии 3.5.1 и понижении до версии 2.7.0 (которая, похоже, является последней версией, которая принимает параметр cpu_shares для create_container) , исправил эту проблему.

Попробуйте запустить это, чтобы понизить версию докерской библиотеки: 

sudo pip3 install docker==2.7.0
...