Воздушный поток: передать {{params.hotel_ids}} в качестве параметра в PostgresOperator

Question

Можем ли мы передать список массивов в параметрах Postflow Airflow?

Я пробовал что-то вроде

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. Это напечатает [1, 2, 3] в операторе SQL, что, конечно, недопустимо SQL.

Airflow использует Jinja для визуализации шаблонов. Чтобы получить правильное выражение SQL, вы можете использовать для циклов в Jinja для рендеринга параметров. Airflow также поддерживает добавление ваших собственных функций для использования в Jinja. Это означает, что вам нужно создать плагин Airflow . Например, вы можете добавить следующий макрос:

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

Если вы правильно импортировали его как плагин, ваш шаблон sql превратится в следующее:

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