I'm trying to set up airflow with two dags on an interval of "every 30 minutes".
The first dag (extract) looks like this:
extract_table_user_schema >> extract_table_user_data >> finish_extract_table_user
The second dag (transform) looks like this:
sensor_wait_for_finish_extract_table_user >> transform_table_user >> finish_transform_table_user
For both dags I've set up the schedule_interval to "*/30 * * * *"
The ExternalTaskSensor is set up with execution_delta=timedelta(minutes=30)
My expected flow of the tasks would be: at first the extract dag is run. Then, after the dummy task finish_tranform_table_user is successful the sensor is triggered and the tasks in transform are run.
It just runs fine for the first time, but when I wait for the second run the tasks interrupt themself.
I've observed the extract dag being interupted by the transform_table_user task which changes some column names. This results in a failed extract_table_user_data since the column names do not match any more.
Edit: to be more specific what are the expected & occurring results
First, the rundown what each task is supposed to do:
(sql)extract_table_user_schema: copies the schema from the user table
(sql)extract_table_user_data: copies data from the source table user into the copy of that table
(dummy)finish_extract_table_user: dummy task
(sensor)sensor_wait_for_finish_extract_table_user: sensor to wait until the extract is finished
(sql)transform_table_user: renames a column field from id to user_id
(dummy)finish_transform_table_user: dummy task
Now, the first time the dags are run the result is a copied schema + data and a renamed field. All tasks as successful
The second time the dags are run, one tasks fails and one task is not run. The other are all successful.
The task extract_table_user_data fails with the error message; no such field as id.
The task finish_extract_table_user was not run
The resulting table was a copied schema but no data and a renamed field.
So I assume the task transform_table_user was run right after the schema was copied. Maybe the set execution_delta=timedelta(minutes=30) is wrong somehow? Maybe I'm misunderstanding the api doc for the sensor?
Help would be appreciated!