如何将过去的TaskInstance的状态设置为成功，并在airflow DAG内继续执行流程？

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

How to set the state of a past TaskInstance to success and continue the pipeline inside a airflow DAG?

问题

  1. 我试图创建一个DAG，其中可以使用dag_id/task_id触发参数。此DAG的目标是将上一个执行任务的状态设置为“success”，并从这一点继续管道。
  3. 管道示例:
  5. [![DAG管道示例][1]][1]
  8.   [1]: https://i.stack.imgur.com/HZc7q.png
  10. 在我的DAG中，我希望能够将“run_that”设置为成功，并自动运行“run_them”作为新状态更改的结果。
  12. 到目前为止，我已经完成了以下工作:
  14. ```py
  15. import airflow
  16. from airflow.models import DagRun, TaskInstance, DagBag
  18. from airflow.operators.dagrun_operator import TriggerDagRunOperator
  19. from airflow.utils.trigger_rule import TriggerRule
  20. from airflow.utils.state import State
  22. from airflow.operators.python_operator import PythonOperator
  24. import pendulum
  26. from wrapper import ScriptLauncher, handleErrorSlack, handleErrorMail
  28. from datetime import timedelta, datetime
  30. default_args = {
  31.   'owner': 'tozzi',
  32.   'depends_on_past': False,
  33.   'start_date': pendulum.datetime(2022, 12, 19, tz='Europe/Paris'),
  34.   'retries': 0,
  35.   'retry_delay': timedelta(minutes=5),
  36.   'xcom_push': True,
  37.   'catchup': False,
  38.   'params': {
  39.     'dag_id': 'my_dag',
  40.     'task_id': 'run_that',
  41.   }
  42. }
  44. def last_exec(dag_id, task_id, session):
  45.     task_instances = (
  46.         session.query(TaskInstance)
  47.         .filter(TaskInstance.dag_id == dag_id, TaskInstance.task_id == task_id)
  48.         .all()
  49.     )
  50.     task_instances.sort(key=lambda x: x.execution_date, reverse=True)
  51.     if task_instances:
  52.         return task_instances[0]
  53.     return None
  55. def set_last_task_success(**kwargs):
  56.     dag_id = kwargs['dag_id']
  57.     task_id = kwargs['task_id']
  59.     session = airflow.settings.Session()
  60.     task_instance = last_exec(dag_id, task_id, session)
  61.     
  62.     if (task_instance is not None):
  63.       task_instance.state = 'success'
  64.       # task_instance = TaskInstance(task_id=task_id, execution_date=last_task_instance.execution_date)
  65.       task_instance.run(session=session, ignore_ti_state=True, ignore_task_deps=True)
  66.       session.commit()
  67.       session.close()
  70. doc_md=f"""## 设置给定dag_id的任务task_id为成功状态"""
  72. # 远程启动
  73. launcher = ScriptLauncher(default_args, "@once", 'set_task_to_success', ['airflow'], doc_md)
  75. dag = launcher.dag;
  77. set_to_success = PythonOperator(
  78.     task_id='set_to_success',
  79.     provide_context=True,
  80.     python_callable=set_last_task_success,
  81.     dag=dag,
  82.     op_kwargs={
  83.         'dag_id': '{{ params.dag_id }}',
  84.         'task_id': '{{ params.task_id }}',
  85.     }
  86. )

这里的task_instance.run(...)调用失败，出现以下错误："AttributeError: 'TaskInstance'对象没有'task'属性"，不过状态更改是正常工作的。我应该更改什么以便在更改“run_that”任务的状态时重新运行“run_them”任务？

  2. <details>
  3. <summary>英文:</summary>
  4. I&#39;m trying to make a DAG with params that can be triggered with a dag_id/task_id. The goal of this DAG is to set the state of the last executed task to &quot;success&quot; and to continue the pipeline from this point.
  5. exemple of pipeline:
  6. [![DAG pipeline exemple][1]][1]
  7. [1]: https://i.stack.imgur.com/HZc7q.png
  8. In my dag I want to be able to set &quot;run_that&quot; to success and automatically run &quot;run_them&quot; as a result of the new state change.
  9. Here is what I did from now:
  10. ```py
  11. import airflow
  12. from airflow.models import DagRun, TaskInstance, DagBag
  13. from airflow.operators.dagrun_operator import TriggerDagRunOperator
  14. from airflow.utils.trigger_rule import TriggerRule
  15. from airflow.utils.state import State
  16. from airflow.operators.python_operator import PythonOperator
  17. import pendulum
  18. from wrapper import ScriptLauncher, handleErrorSlack, handleErrorMail
  19. from datetime import timedelta, datetime
  20. default_args = {
  21. &#39;owner&#39;: &#39;tozzi&#39;,
  22. &#39;depends_on_past&#39;: False,
  23. &#39;start_date&#39;: pendulum.datetime(2022, 12, 19, tz=&#39;Europe/Paris&#39;),
  24. &#39;retries&#39;: 0,
  25. &#39;retry_delay&#39;: timedelta(minutes=5),
  26. &#39;xcom_push&#39;: True,
  27. &#39;catchup&#39;: False,
  28. &#39;params&#39;: {
  29. &#39;dag_id&#39;: &#39;my_dag&#39;,
  30. &#39;task_id&#39;: &#39;run_that&#39;,
  31. }
  32. }
  33. def last_exec(dag_id, task_id, session):
  34. task_instances = (
  35. session.query(TaskInstance)
  36. .filter(TaskInstance.dag_id == dag_id, TaskInstance.task_id == task_id)
  37. .all()
  38. )
  39. task_instances.sort(key=lambda x: x.execution_date, reverse=True)
  40. if task_instances:
  41. return task_instances[0]
  42. return None
  43. def set_last_task_success(**kwargs):
  44. dag_id = kwargs[&#39;dag_id&#39;]
  45. task_id = kwargs[&#39;task_id&#39;]
  46. session = airflow.settings.Session()
  47. task_instance = last_exec(dag_id, task_id, session)
  48. if (task_instance is not None):
  49. task_instance.state = &#39;success&#39;
  50. # task_instance = TaskInstance(task_id=task_id, execution_date=last_task_instance.execution_date)
  51. task_instance.run(session=session, ignore_ti_state=True, ignore_task_deps=True)
  52. session.commit()
  53. session.close()
  54. doc_md=f&quot;&quot;&quot;## Set the given task_id to success of the given dag_id&quot;&quot;&quot;
  55. # launched remotely
  56. launcher = ScriptLauncher(default_args, &quot;@once&quot;, &#39;set_task_to_success&#39;, [&#39;airflow&#39;], doc_md)
  57. dag = launcher.dag;
  58. set_to_success = PythonOperator(
  59. task_id=&#39;set_to_success&#39;,
  60. provide_context=True,
  61. python_callable=set_last_task_success,
  62. dag=dag,
  63. op_kwargs={
  64. &#39;dag_id&#39;: &#39;{{ params.dag_id }}&#39;,
  65. &#39;task_id&#39;: &#39;{{ params.task_id }}&#39;,
  66. }
  67. )

The task_instance.run(...) call fail here with this error : "AttributeError: 'TaskInstance' object has no attribute 'task'", the state change is correctly working tho. What should I change so it rerun the "run_them" task when I change the state of the "run_that" task?

答案1

得分: 2

截至Airflow 2.5.0版本，您可以通过Airflow REST API更新任何现有任务的状态。

英文:

As of Airflow 2.5.0 you can update the state of any existing task via the Airflow REST API.

答案2

得分: 1

使用@TJaniF的答案，我制作了这个小型可重用的失败函数任务on_failure_send_force_success_mail，该函数会发送一封邮件，其中包含指向自定义API的链接，该API调用使用GET请求来调用修补任务实例请求，它的运行符合预期：

  1. def on_failure_send_force_success_mail(task_id):
  2.     def send_mail(**context):
  3.         result = requests.post('mailing_api_url', json={
  4.             "from": "foobar@mail.com",
  5.             "to": "tozzi@mail.com",
  6.             "subject": "Airflow task failed",
  7.             "html": f'Airflow task test failed: <a href="custom_api_url/script/{context["task_instance"].dag_id}/dag_runs/{context["dag_run"].run_id}/task_instances/{task_id}?new_state=success">click here</a>',
  8.         })
  9.     return send_mail
  11. def run_this_func_failed(**context):
  12.     raise Exception('FAILED')
  14. def run_this_func(**context):
  15.     print('COMPLETED')
  17. run_that = PythonOperator(
  18.     task_id='run_that',
  19.     provide_context=True,
  20.     python_callable=run_this_func_failed,
  21.     dag=dag,
  22. )
  24. run_this = PythonOperator(
  25.     task_id='run_this',
  26.     provide_context=True,
  27.     python_callable=run_this_func,
  28.     trigger_rule="all_success",
  29.     dag=dag,
  30. )
  32. on_failure = PythonOperator(
  33.     task_id='on_failure',
  34.     provide_context=True,
  35.     python_callable=on_failure_send_force_success_mail('run_that'),
  36.     trigger_rule="all_failed",
  37.     dag=dag,
  38. )
  40. run_that >> on_failure
  41. run_that >> run_this
英文:

Using @TJaniF answer, I made this little reusable failure function task on_failure_send_force_success_mail, this function send a mail with a link to a custom API that call the patch task instance request using a get, it works as expected:

  1. def on_failure_send_force_success_mail(task_id):
  2.     def send_mail(**context):
  3.         result = requests.post(&#39;mailing_api_url&#39;, json={
  4.             &quot;from&quot;: &quot;foobar@mail.com&quot;,
  5.             &quot;to&quot;: &quot;tozzi@mail.com&quot;,
  6.             &quot;subject&quot;: &quot;Airflow task failed&quot;,
  7.             &quot;html&quot;: f&#39;Airflow task test failed: &lt;a href=&quot;custom_api_url/script/{context[&quot;task_instance&quot;].dag_id}/dag_runs/{context[&quot;dag_run&quot;].run_id}/task_instances/{task_id}?new_state=success&quot;&gt;click here&lt;/a&gt;&#39;,
  8.         })
  9.     return send_mail
  11. def run_this_func_failed(**context):
  12.     raise Exception(&#39;FAILED&#39;)
  14. def run_this_func(**context):
  15.     print(&#39;COMPLETED&#39;)
  17. run_that = PythonOperator(
  18.     task_id=&#39;run_that&#39;,
  19.     provide_context=True,
  20.     python_callable=run_this_func_failed,
  21.     dag=dag,
  22. )
  24. run_this = PythonOperator(
  25.     task_id=&#39;run_this&#39;,
  26.     provide_context=True,
  27.     python_callable=run_this_func,
  28.     trigger_rule=&quot;all_success&quot;,
  29.     dag=dag,
  30. )
  32. on_failure = PythonOperator(
  33.     task_id=&#39;on_failure&#39;,
  34.     provide_context=True,
  35.     python_callable=on_failure_send_force_success_mail(&#39;run_that&#39;),
  36.     trigger_rule=&quot;all_failed&quot;,
  37.     dag=dag,
  38. )
  40. run_that &gt;&gt; on_failure
  41. run_that &gt;&gt; run_this

huangapple
  • 本文由 发表于 2023年2月6日 21:18:16
  • 转载请务必保留本文链接：https://go.coder-hub.com/75361826.html
  • airflow
  • python
匿名

发表评论

匿名网友

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

确定