airflow - 如何通过气流将参数传递给 hql 运行

Question

我想知道如何将参数传递给通过气流运行的配置单元查询脚本。如果我只想为此脚本添加一个参数，比如 target_db = mydatabase，我该怎么做？我是否需要将它添加到 default_args 然后调用它然后在脚本的 op_kwargs 中调用它？

default_args = {
    'owner': 'airflow',
    'depends_on_past': True,
    'start_date': datetime(2017, 11, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(dag_name, default_args=default_args, schedule_interval="@daily")

t_add_step = PythonOperator(
    task_id='add__step',
    provide_context=True,
    python_callable=add_emr_step,
    op_kwargs={
        'aws_conn_id': dag_params['aws_conn_id'],
        'create_job_flow_task': 'create_emr_flow',
        'get_step_task': 'get_email_step'
    },
    dag=dag
)

Answer 1

假设您正在使用 BashOperator 调用 Hive，它看起来像这样

...
set_hive_db = BashOperator (
    bash_command = """
     hive --database {{params.database}} -f {{params.hql_file}}
     """,
     params = {
       "database": "testingdb",
       "hql_file": "myhql.hql"
     },
     dag = dag
)
...

另一种方法是USE database进入你的 hql 并调用hive -f hqlfile.hql你的 BashOperator