`airflow.exceptions.AirflowException: ‘branch_task_ids’必须只包含有效的task_ids`

huangapple
go评论95阅读模式
英文:

airflow.exceptions.AirflowException: 'branch_task_ids' must contain only valid task_ids

问题

  1. 我有一个DAG，其中包含1个自定义任务，1个`@task.branch`任务修饰符和1个`taskgroup`，在taskgroup中，有多个任务需要根据`@task.branch`的结果依次触发。
  3. PROCESS_BATCH_data_FILE = "batch_upload"
  4. SINGLE_data_FILE_FIRST_OPERATOR = "validate_data_schema_task"
  5. ENSURE_INTEGRITY_TASK = "provide_data_integrity_task"
  6. PROCESS_SINGLE_data_FILE = "process_single_data_file_task"
  8. default_args = {
  9.     "retries": 0,
  10.     "retry_delay": timedelta(seconds=30),
  11.     "trigger_rule": "none_failed",
  12. }
  14. default_args = update_default_args(default_args)
  16. flow_name = "data_ingestion"
  18. with DAG(
  19.     flow_name,
  20.     default_args=default_args,
  21.     start_date= airflow.utils.dates.days_ago(0),
  22.     schedule=None,
  23.     dagrun_timeout=timedelta(minutes=180)
  24. ) as dag:
  26.     update_status_running_op = UpdateStatusOperator(
  27.         task_id="update_status_running_task",
  28.     )
  30.     @task.branch(task_id='check_payload_type')
  31.     def is_batch(**context):
  33.         # data = context["dag_run"].conf["execution_context"].get("data")
  35.         if isinstance(data, dict):
  36.             subdag = "validate_data_schema_task"
  37.         elif isinstance(data, list):
  38.             subdag = PROCESS_BATCH_data_FILE
  39.         return subdag
  41.     with TaskGroup(group_id='group1') as my_task_group:
  42.         validate_schema_operator = ValidatedataSchemaOperator(task_id=SINGLE_data_FILE_FIRST_OPERATOR)
  43.         ensure_integrity_op = EnsuredataIntegrityOperator(task_id=ENSURE_INTEGRITY_TASK)
  44.         process_single_data_file = ProcessdataOperatorR3(task_id=PROCESS_SINGLE_data_FILE)
  46.         validate_schema_operator >> ensure_integrity_op >> process_single_data_file 
  48.     update_status_finished_op = UpdateStatusOperator(
  49.         task_id="update_status_finished_task",
  50.         dag=dag,
  51.         trigger_rule="all_done",
  52.     )
  54.     batch_upload = DummyOperator(
  55.         task_id=PROCESS_BATCH_data_FILE
  56.     )
  58.     for batch in range(0, BATCH_NUMBER):
  59.         batch_upload >> ProcessdataOperatorR3(
  60.             task_id=f"process_data_task_{batch + 1}",
  61.             previous_task_id=f"provide_data_integrity_task_{batch + 1}",
  62.             batch_number=batch + 1,
  63.             trigger_rule="none_failed_or_skipped"
  64.         ) >> update_status_finished_op
  66. branch_task = is_batch()
  68. update_status_running_op >> branch_task
  69. branch_task >> batch_upload
  70. branch_task >> my_task_group >> update_status_finished_op
英文:

I have a dag which contains 1 custom task, 1 @task.branch task decorator and 1 taskgroup, inside the taskgroup I have multiple tasks that need to be triggered sequentially depending on the outcome of the @task.branch.

  1. PROCESS_BATCH_data_FILE = "batch_upload"
  2. SINGLE_data_FILE_FIRST_OPERATOR = "validate_data_schema_task"
  3. ENSURE_INTEGRITY_TASK = "provide_data_integrity_task"
  4. PROCESS_SINGLE_data_FILE = "process_single_data_file_task"
  7. default_args = {       
  8.     "retries": 0,
  9.     "retry_delay": timedelta(seconds=30),
  10.     "trigger_rule": "none_failed",
  11. }
  13. default_args = update_default_args(default_args)
  15. flow_name = "data_ingestion"
  17. with DAG(
  18.     flow_name,
  19.     default_args=default_args,
  20.     start_date= airflow.utils.dates.days_ago(0),
  21.     schedule=None,
  22.     dagrun_timeout=timedelta(minutes=180)
  23. ) as dag:
  25.     update_status_running_op = UpdateStatusOperator(
  26.         task_id="update_status_running_task",
  27.     )
  29.     @task.branch(task_id = 'check_payload_type')
  30.     def is_batch(**context):
  32.         # data = context["dag_run"].conf["execution_context"].get("data")
  34.         if isinstance(data, dict):
  35.             subdag = "validate_data_schema_task"
  36.         elif isinstance(data, list):
  37.             subdag = PROCESS_BATCH_data_FILE
  38.         return subdag
  41.     with TaskGroup(group_id='group1') as my_task_group:
  42.         validate_schema_operator = ValidatedataSchemaOperator(task_id=SINGLE_data_FILE_FIRST_OPERATOR)
  43.         ensure_integrity_op = EnsuredataIntegrityOperator(task_id=ENSURE_INTEGRITY_TASK)
  44.         process_single_data_file = ProcessdataOperatorR3(task_id=PROCESS_SINGLE_data_FILE)
  47.         validate_schema_operator >> ensure_integrity_op >> process_single_data_file 
  50.     update_status_finished_op = UpdateStatusOperator(
  51.         task_id="update_status_finished_task",
  52.         dag=dag,
  53.         trigger_rule="all_done",
  54.     )
  57.     batch_upload = DummyOperator(
  58.         task_id=PROCESS_BATCH_data_FILE
  59.     )
  61.     for batch in range(0, BATCH_NUMBER):
  62.         batch_upload >> ProcessdataOperatorR3(
  63.             task_id=f"process_data_task_{batch + 1}",
  64.             previous_task_id=f"provide_data_integrity_task_{batch + 1}",
  65.             batch_number=batch + 1,
  66.             trigger_rule="none_failed_or_skipped"
  67.         ) >> update_status_finished_op
  69. branch_task = is_batch()
  71. update_status_running_op >> branch_task
  72. branch_task >> batch_upload
  73. branch_task >> my_task_group >> update_status_finished_op

When I trigger below tag I get the following error:

  1. airflow.exceptions.AirflowException: 'branch_task_ids' must contain only valid task_ids. Invalid tasks found: {'validate_data_schema_task'}.

I dont understand why I get the error because 'validate_data_schema_task' is defined at the top of the file. I have tried to hard code 'validate_data_schema_task' as task_id but that gives me the same error.

答案1

得分: 1

任务组中的任务默认情况下在其task_id前面加上了任务组的group_id。这意味着，由于您的ValidatedataSchemaOperator任务位于“group1”的任务组中，该任务的task_id实际上是“group1.validate_data_schema_task”。添加group_id的目的是最初确保DAG中任务的唯一性。

如果希望禁用从TaskGroup添加group_id到底层任务中，可以在TaskGroup上设置prefix_group_id=False。

在Airflow文档的TaskGroups部分中有一些其他细节。

英文:

Tasks within TaskGroups by default have the TaskGroup's group_id prepended to the task_id. Meaning since your ValidatedataSchemaOperator task is in a TaskGroup of "group1", that task's task_id is actually "group1.validate_data_schema_task". The prepending of the group_id is to initially ensure uniqueness of tasks within a DAG.

You can disable adding the group_id from TaskGroup to the underlying tasks by setting prefix_group_id=False on the TaskGroup if you'd like.

There are some other finer-grain details on TaskGroups in the Airflow docs.

答案2

得分: 1

当涉及到嵌套在任务组中的任务时，你需要指定它们的任务 _id 为 "group_id.task_id"。

这应该可以工作：

  1.     @task.branch(task_id='check_payload_type')
  2.     def is_batch(**context):
  4.         # data = context["dag_run"].conf["execution_context"].get("data")
  6.         if isinstance(data, dict):
  7.             subdag = "group1.validate_data_schema_task"
  8.         elif isinstance(data, list):
  9.             subdag = "group1." + PROCESS_BATCH_data_FILE
  10.         return subdag
英文:

When referring to a task nested in a task group you need to specify their task _id as "group_id.task_id".

This should work:

  1.     @task.branch(task_id = 'check_payload_type')
  2.     def is_batch(**context):
  4.         # data = context["dag_run"].conf["execution_context"].get("data")
  6.         if isinstance(data, dict):
  7.             subdag = "group1.validate_data_schema_task"
  8.         elif isinstance(data, list):
  9.             subdag = "group1." + PROCESS_BATCH_data_FILE
  10.         return subdag
  11.

huangapple
  • 本文由 发表于 2023年3月7日 22:59:51
  • 转载请务必保留本文链接：https://go.coder-hub.com/75663632.html
  • airflow
  • airflow-taskflow
  • python
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定