Airflow DAG - 如何先检查 BQ（必要时删除）然后运行数据流作业？

Question

我正在使用 Cloud Composer 为到达 GCS 并进入 BigQuery 的文件编排 ETL。我有一个云函数，当文件到达时触发 dag，云函数将文件名/位置传递给 DAG。在我的 DAG 中，我有 2 个任务：

1) 使用DataflowPythonOperator 运行数据流作业，从 GCS 中的文本中读取数据并将其转换并输入到 BQ，然后 2) 移动文件到失败/成功存储桶，具体取决于作业是失败还是成功。 每个文件都有一个文件 ID，它是 bigquery 表中的一列。有时一个文件会被编辑一次或两次（这不是经常发生的流式传输），我希望能够首先删除该文件的现有记录。

我查看了其他气流运算符，但在运行数据流作业之前希望在我的 DAG 中有 2 个任务：

1. 根据文件名获取文件 ID（现在我有一个 bigquery 表映射文件名 -> 文件 ID，但我也可以只引入一个用作地图的 json，我想这是否更容易)
2. 如果文件 ID 已经存在于 bigquery 表（从数据流作业输出转换后的数据的表）中，请将其删除，然后运行数据流作业，这样我就有了最新信息。我知道一个选择是只添加一个时间戳并且只使用最新的记录，但是因为每个文件可能有 100 万条记录，这不像我每天删除 100 个文件（可能是 1-2 个顶部）这似乎是混乱和混乱。

在数据流作业之后，理想情况下，在将文件移动到成功/失败文件夹之前，我想附加一些“记录”表，说明此时输入了该游戏。这将是我查看发生的所有插入的方式。 我试图寻找不同的方法来做到这一点，我是云作曲家的新手，所以在经过 10 多个小时的研究后，我不清楚这将如何工作，否则我会发布代码以供输入。

谢谢，我非常感谢大家的帮助，如果这不是你想要的那么清楚，我深表歉意，关于气流的文档非常强大，但考虑到云作曲家和 bigquery 相对较新，很难彻底了解如何做一些 GCP 特定的任务。

Answer 1

听起来有点复杂。很高兴，几乎所有 GCP 服务都有运营商。另一件事是何时触发 DAG 执行。你想通了吗？您希望每次有新文件进入该 GCS 存储桶时触发 Google Cloud Function 运行。

1. 触发 DAG

要触发 DAG，您需要使用依赖于 Object Finalize 或 Metadata Update 触发器的 Google Cloud 函数来调用它。

1. 将数据加载到 BigQuery

如果您的文件已经采用 GCS 格式，并且采用 JSON 或 CSV 格式，那么使用 Dataflow 作业就显得多余了。您可以使用GoogleCloudStorageToBigQueryOperator 将文件加载到BQ。

1. 跟踪文件 ID

计算文件 ID 的最佳方法可能是使用来自 Airflow 的 Bash 或 Python 运算符。你能直接从文件名中推导出来吗？

如果是这样，那么您可以在GoogleCloudStorageObjectSensor 的上游使用 Python 运算符来检查文件是否在成功的目录中。

如果是，那么您可以使用BigQueryOperator 在 BQ 上运行删除查询。

之后，您运行 GoogleCloudStorageToBigQueryOperator。

1. 移动文件

如果您要将文件从 GCS 移动到 GCS 位置，那么 GoogleCloudStorageToGoogleCloudStorageOperator 应该可以满足您的需要。如果您的 BQ 加载运算符失败，则移动到失败的文件位置，如果成功，则移动到成功的作业位置。

4. 记录任务日志

也许您需要跟踪插入的只是将任务信息记录到 GCS。查看how to log task information to GCS

这有帮助吗？