Probleme mit Airflow DAG

Question

Ich arbeite seit einiger Zeit an Apache Airflow, um meinen Workflow zu planen. Ich habe Probleme mit der Planung meiner DAG. Ich habe mit dieser SO Frage Referenz: 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) 

Ich versuche, die Aufgabe mit der folgenden Liste von Befehlen auszuführen:

airflow scheduler 
airflow trigger_dag dag_mkdir_folder 

Ich halte dies als Fehler bekommen:

 
[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 

Der Bash-Befehl soll nur ein neues Verzeichnis erstellen. Die Testversion funktioniert gut.

Answer 1

Führen Sie den Scheduler auf einem anderen Terminal und dann auslösen Ihre dag in einem anderen Terminal

auch versuchen, Ihren vollständigen Pfad bereitstellt, wo Sie das Verzeichnis machen wollen. Zum Beispiel Ordner, in dem Luftstrom-Verzeichnis zu erstellen:

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

Dies sollte test_airflow Ordner im Luftstrom

Answer 2

Ihre aktuelle bash_command sagt Airflow erstellen Sie ein Verzeichnis innerhalb des temporären Verzeichnisses, das die DAG verwendet erstellen, wenn es running, das er mit all seinen Inhalten wegbläst, nachdem die DAG ausgeführt wurde.

Betrachten Sie in das Verzeichnis zu ändern Sie es in erstellen möchten

A bash_command mit, dass so etwas aussehen würde.

"cd <path_to_directory>; mkdir test_airflow"