Airflow 中的 BigQuery 运算符未将 sql 作为原始文件读取答案

【问题标题】：BigQuery Operator in Airflow not reading sql as raw fileAirflow 中的 BigQuery 运算符未将 sql 作为原始文件读取
【发布时间】：2021-03-17 05:58:01
【问题描述】：

我的目标是在 GCS 中存储 sql 代码，并使用 Bigquery Operator 传递给我的日常 ETL 流程的“sql”参数。

我的错误是

Exception: BigQuery job failed. Final error was: {'reason': 'invalidQuery', 'location': 'query', 'message': 'Syntax error: Expected end of input but got "/" at [1:1]'}

我目前遇到的问题是sql代码，很简单

Select * 
FROM 
table

将换行符 /n 作为 sql 文件的一部分读取，因此它作为无效查询失败。当我在python中读取sql文件时，读取如下

'SELECT * \nFROM \n    `table` as test\n'

我的任务如下

t1 = BigQueryOperator(
  task_id='bigquery_test',
  sql= '/home/airflow/.../filename.sql ',
  destination_dataset_table = destination_table,
  write_disposition='WRITE_TRUNCATE',
  use_legacy_sql=False,
  dag=dag,
)

我假设这与 sql 参数在气流中的设置方式有关，但我无法完全理解解决此问题所需要做的事情。

非常感谢任何帮助。

【问题讨论】：

标签： python google-bigquery airflow

【解决方案1】：

为了进一步为社区做出贡献，我对@Pentium10的回答进行补充和补充。

根据documentation，sql 参数应该接收代表 sql 语句的字符串或带有您要运行的查询的 .sql 模板文件。此文件应存储在文件夹 /home/airflow/gcs/data 中，因为此文件夹存储任务生成和使用的数据。此外，它安装在所有工作节点上，here。因此，您应该引用您的模板化 .sql 文件（不是整个路径），如下所示：

t1 = BigQueryOperator(
  task_id='bigquery_test',
  sql= 'filename.sql',
  destination_dataset_table = destination_table,
  write_disposition='WRITE_TRUNCATE',
  use_legacy_sql=False,
  dag=dag,
)

另外，我必须指出，如果您想引用另一个文件夹中的模板化sql 文件，例如/home/airflow/gcs/data 中的子文件夹，您必须指定template_searchpath 参数。它指定 Jinja 将在其中查找模板的文件夹 link。下面是它的语法，

与模型.DAG( 'bigqueryoperator_dag', template_searchpath='/home/airflow/gcs/data/bigqueryoperator/', default_args=DEFAULT_DAG_ARGS) 作为 dag：

bqo = BigQueryOperator(
    task_id='bigqueryoperator_templated_sql',
    sql='sample_query_tempalte.sql',
    use_legacy_sql=False,
)

请注意，文件 sample_query_tempalte.sql 位于子文件夹 /home/airflow/gcs/data/bigqueryoperator/ 中。

【讨论】：

@Tristan，如果您发现此信息有用。请考虑接受并支持答案。
非常有帮助！非常感谢

【解决方案2】：

预期输入结束，但在 [1:1] 处得到“/”

你的问题是第 3 行：

它说第一个字符是/，这是你路径的第一个字符。

t1 = BigQueryOperator(
  task_id='bigquery_test',
  sql= '/home/airflow/.../filename.sql ',

而不是传递一个路径字符串，你需要读取文件的内容，并传递文件的内容。

【讨论】：