气流：将 {{ params.hotel_ids }} 作为参数传递给 PostgresOperator

Question

我们可以在 Airflow postgress 参数中传递数组列表吗？

我正在尝试类似的东西

Select hotel_name from hoteldetails where id in ({{ params.hotel_ids }});

在“s3_to_redshift.sql”中。

下面是我的运营商::

s3_to_redshift = PostgresOperator(
    task_id='s3_to_redshift',
    postgres_conn_id='redshift',
    sql='s3_to_redshift.sql',
    params={'hotel_ids': [1,2,3]},
    dag=dag
)

Answer 1

你可以，但你必须担心你的参数将如何呈现。在这种情况下，您将传递数组并直接在 SQL 语句中呈现它。这将在 SQL 语句中打印[1, 2, 3]，这当然是无效的 SQL。

Airflow 使用 Jinja 来渲染模板。为了获得正确的 SQL 语句，您可以在 Jinja 中使用 for loops 来呈现参数。 Airflow 还支持添加您自己的函数以在 Jinja 中使用。这意味着您需要创建一个Airflow plugin。例如，您可以添加以下宏：

def render_list_sql(list):
  return ', '.join(list)

如果您将其作为插件正确导入，您的 sql 模板将变为以下内容：

Select hotel_name from hoteldetails where id in ({{ macros.render_list_sql(params.hotel_ids) }});