AWS Airflow配置JSON触发器

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

AWS Airflow trigger with configuration JSON

问题

我尝试使用Airflow创建动态触发器，也就是说，在触发Airflow时，我想在触发脚本之前更改多个参数。目前，我卡在如何使用AWS Airflow提供的配置JSON界面传递参数上。

Airflow脚本如下：

  1. from airflow import DAG
  2. from airflow.operators.python_operator import PythonOperator
  3. from datetime import datetime
  5. def print_hello():
  6.     message = 'hello world'
  7.     print(message)
  9. default_args = {
  10.     'owner': 'airflow',
  11.     'depends_on_past': False,
  12.     'start_date': datetime(2023, 8, 7),
  13.     'retries': 1,
  14. }
  16. dag = DAG(
  17.     'hello_world_dag',
  18.     default_args=default_args,
  19.     schedule_interval=None,
  20.     catchup=False,
  21. )
  23. print_hello_task = PythonOperator(
  24.     task_id='print_hello_task',
  25.     python_callable=print_hello,
  26.     dag=dag,
  27. )
  29. print_hello_task

在这个示例中，我将参数传递到配置JSON中。

如何将参数“message”从配置JSON传递到脚本中？

英文:

i try to make dynamic trigger with airflow, means during trigger airflow i want to change multiple parameter before i trigger the scripts.. right now i stuck on how to pass the parameter using configuration json UI provide by aws airflow.

airflow scripts as below :

  1.     from airflow import DAG
  2. from airflow.operators.python_operator import PythonOperator
  3. from datetime import datetime
  5. def print_hello():
  6.     message = 'hello world'
  7.     print(message)
  9. default_args = {
  10.     'owner': 'airflow',
  11.     'depends_on_past': False,
  12.     'start_date': datetime(2023, 8, 7),
  13.     'retries': 1,
  14. }
  16. dag = DAG(
  17.     'hello_world_dag',
  18.     default_args=default_args,
  19.     schedule_interval=None,
  20.     catchup=False,
  21. )
  23. print_hello_task = PythonOperator(
  24.     task_id='print_hello_task',
  25.     python_callable=print_hello,
  26.     dag=dag,
  27. )
  29. print_hello_task

this example i pass the parameter into configuration json.
AWS Airflow配置JSON触发器

how i can pass parameter "message" from configuration json to the scripts?

答案1

得分: 2

在此情况下，非常重要将 render_template_as_native_obj=True 添加到Dag定义中，因为您希望将此配置视为字典。

配置将作为kwargs传递给函数

  1. with DAG(
  2.     dag_id="hello_world_dag",
  3.     schedule_interval=None,
  4.     default_args=default_args,
  5.     render_template_as_native_obj=True,
  6.     catchup=False,
  7. ) as dag:
  9.     def print_hello(**kwargs):
  10.         print(kwargs.get("message"))
  12.     print_hello_task = PythonOperator(
  13.         task_id='print_hello_task',
  14.         python_callable=print_hello,
  15.         op_kwargs="{{ dag_run.conf }}",
  16.     )
  18.     print_hello_task
英文:

Another option in to use Jinja Template {{ dag_run.conf }} as mention in the ui

in this case very important to add render_template_as_native_obj=True to the Dag definition cause you want to consider this conf as dict.

the conf would be passed to the function as kwargs

  1. with DAG(
  2.     dag_id="hello_world_dag",
  3.     schedule_interval=None,
  4.     default_args=default_args,
  5.     render_template_as_native_obj=True,
  6.     catchup=False,) as dag:
  8.     def print_hello(**kwargs):
  9.         print(kwargs.get("message"))
  12.     print_hello_task = PythonOperator(
  13.         task_id='print_hello_task',
  14.         python_callable=print_hello,
  15.         op_kwargs="{{ dag_run.conf }}",
  16.     )
  18.     print_hello_task

答案2

得分: 1

以下是您要翻译的代码部分：

  1. 这是一个示例DAG，它使用Airflow的TaskFlow API打印提供的DAG运行配置对象`{"hello": "world"}`的一些属性（请参阅[上下文](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/taskflow.html#context)部分以获取有关访问任务实例上下文的更多信息）：
  3. from pendulum import datetime
  5. from airflow.decorators import (
  6.     dag,
  7.     task,
  8. )
  10. @dag(
  11.     schedule="@daily",
  12.     start_date=datetime(2023, 1, 1),
  13.     catchup=False,
  14.     default_args={
  15.         "retries": 2,
  16.     },
  17.     tags=["example"],
  18. )
  19. def read_config():
  20.     @task
  21.     def read_conf(dag_run=None):
  22.         print(type(dag_run.conf))
  23.         print(f"DAG Run config: {dag_run.conf}")
  24.         print(f"Hello {dag_run.conf['hello']}")
  26.     read_conf()
  28. read_config()

任务实例日志：

  1. [2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - <class 'dict'>
  2. [2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - DAG Run config: {'hello': 'world'}
  3. [2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - Hello world
英文:

Here's an example DAG that prints some properties of a provided dag run config object of {&quot;hello&quot;: &quot;world&quot;} using Airflow's TaskFlow API (see the context section for more info on accessing a task instance's context):

  1. from pendulum import datetime
  3. from airflow.decorators import (
  4.     dag,
  5.     task,
  6. )
  9. @dag(
  10.     schedule=&quot;@daily&quot;,
  11.     start_date=datetime(2023, 1, 1),
  12.     catchup=False,
  13.     default_args={
  14.         &quot;retries&quot;: 2,
  15.     },
  16.     tags=[&quot;example&quot;],
  17. )
  18. def read_config():
  19.     @task
  20.     def read_conf(dag_run=None):
  21.         print(type(dag_run.conf))
  22.         print(f&quot;DAG Run config: {dag_run.conf}&quot;)
  23.         print(f&quot;Hello {dag_run.conf[&#39;hello&#39;]}&quot;)
  25.     read_conf()
  28. read_config()

Task instance logs:

  1. [2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - &lt;class &#39;dict&#39;&gt;
  2. [2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - DAG Run config: {&#39;hello&#39;: &#39;world&#39;}
  3. [2023-08-08, 22:35:48 EDT] {logging_mixin.py:150} INFO - Hello world

huangapple
  • 本文由 发表于 2023年8月9日 09:35:25
  • 转载请务必保留本文链接：https://go.coder-hub.com/76864037-2.html
  • airflow
  • python
匿名

发表评论

匿名网友

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

确定