공기 흐름 실행 문제 DAG

Question

지금까지 작업 흐름을 예약하기 위해 Apache Airflow를 한동안 사용해 왔습니다. 내 DAG 스케줄링 문제가있는 것 같습니다. 나는 참조 할 수 있도록이 SO 질문을 사용하고있다 : 나는 오류로이가 계속

airflow scheduler 
airflow trigger_dag dag_mkdir_folder 

:

Airflow not scheduling Correctly Python

from airflow import DAG 
from airflow.operators.bash_operator import BashOperator 
from datetime import datetime 
from datetime import timedelta 

default_args = { 
    'owner': 'airflow', 
    'depends_on_past': False, 
'start_date': datetime.now() - timedelta(minutes=5), 
'email': ['airflow@airflow.com'], 
'email_on_failure': False, 
'email_on_retry': False, 
'retries': 1, 
'retry_delay': timedelta(minutes=5), 
} 

dag = DAG('dag_mkdir_folder', default_args=default_args, 
     schedule_interval=timedelta(minutes=5)) 


task_hello = BashOperator(task_id='print_hello', 
         bash_command='mkdir test_airflow', dag=dag) 

내가 명령의 다음 목록을 사용하여 작업을 실행하기 위해 노력하고있어

 
[2017-05-15 13:49:06,688] {models.py:322} DagFileProcessor406 INFO -  Finding 'running' jobs without a recent heartbeat 
[2017-05-15 13:49:06,689] {models.py:328} DagFileProcessor406 INFO - Failing jobs without heartbeat after 2017-05-15 13:44:06.689284 

bash 명령은 새로운 디렉토리를 생성하기로되어 있습니다. 테스트 버전이 잘 작동합니다.

Answer 1

또한 디렉토리를 만들고 싶어 어디 전체 경로를 제공하려고 다른 터미널

에 DAG를 실행 한 후 다른 터미널에서 스케줄러를 실행합니다. 공기 흐름 디렉토리에 폴더를 생성 예를 들어 : 이것은 공기 흐름 내부 test_airflow 폴더를 만들어야합니다

task_hello = BashOperator(task_id='print_hello', 
        bash_command="mkdir ~/airflow/test_airflow", dag=dag) 

Answer 2

현재 bash_command는이 때 DAG가 사용하는 임시 디렉토리 안에 디렉토리를 생성하기 위해 공기 흐름을 말하고있다 DAG가 실행 된 후 모든 내용물과 함께 사라집니다.

당신이 그것을 만들 디렉토리로 디렉토리를 변경하는 것이 좋습니다 그와

bash_command를 같은 것을 보일 것이다.

"cd <path_to_directory>; mkdir test_airflow" 

당신이 "공기 흐름 스케줄러"를 입력 한 후 무엇을 보았는가