如何使用 Airflow 在 GCS 存储桶中写入文件

Question

这里是 Airflow 的新手。

我有一个 Python 代码，它读取 BigQuery 表，将一些转换为 pandas DataFrame 并将其保存为文件。

使用 Airflow，我需要一个 DAG 来执行我的代码并将其作为文件保存在 Google Cloud Storage 存储桶中。

Airflow 部署在 Composer 上。

我该怎么做？

Answer 1

如果您的转换可以用 BigQuery QL 表示，您可以使用 BQ 到 GCS 运算符：

https://airflow.apache.org/docs/apache-airflow-providers-google/stable/_api/airflow/providers/google/cloud/transfers/bigquery_to_gcs/index.html

这里的例子：

https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/example_dags/example_bigquery_to_gcs.py

如果您需要进行更复杂的转换，而您没有可以编排的外部服务，请创建一个使用 BigQuery Hook 和 GCS Hook 并执行您想做的事情的自定义运算符。它比您想象的要容易 - 只需看一下 BQToGCS 运算符，您就会发现它相当简单。

https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/transfers/bigquery_to_gcs.py

Airflow 完全是 Python - 因此，如果您在 DAG 中组合现有运算符，或者编写自己的运算符（然后组合它们），它并不会真正改变太多。都是python代码。 Airflow 实现了 Hook 抽象，特别是为了能够隐藏与服务通信的复杂性，但允许您作为 DAG/Operator 的编写者使用 hooks 编写 operator 的代码并执行一些额外的操作。