我正在使用云编写器来协调ETL,以便将到达GCS的文件转到BigQuery。我有一个云函数,当文件到达时触发dag并且云函数将文件名/位置传递给DAG。在我的DAG中,我有两个任务:
1)使用DataflowPythonOperator
运行数据流作业,该作业从GCS中的文本读取数据并对其进行转换并将其输入BQ,以及2)根据作业是失败还是成功将文件移动到失败/成功桶。每个文件都有一个文件ID,它是bigquery表中的一列。有时文件将被编辑一次或两次(它不是经常出现的流式传输),我希望能够首先删除该文件的现有记录。
我调查了其他气流操作员,但是在运行数据流作业之前,我希望在DAG中有2个任务:
在数据流作业之后,理想情况下,在将文件移动到成功/失败文件夹之前,我想附加一些“记录”表,说这个游戏是在此时输入的。这将是我查看发生的所有插入的方法。我试图寻找不同的方法来做到这一点,我是云作曲家的新手,所以我没有清楚地知道在经过10多个小时的研究后这将如何工作,否则我会发布代码进行输入。
谢谢,我非常感谢大家的帮助并道歉,如果这不是你想要的那么清楚,关于气流的文档是非常强大的,但鉴于云作曲家和bigquery相对较新,很难学会如何彻底地做一些GCP的具体任务。
听起来有点复杂。很高兴,每个GCP服务都有运营商。另一件事是何时触发DAG执行。你有没有意识到这一点?每当新文件进入该GCS存储桶时,您都希望触发Google Cloud功能。
要触发DAG,您需要使用依赖于Object Finalize或Metadata Update触发器的Google Cloud功能来调用它。
如果您的文件已经在GCS中,并且采用JSON或CSV格式,那么使用Dataflow作业就太过分了。您可以使用GoogleCloudStorageToBigQueryOperator将文件加载到BQ。
计算文件ID的最好的事情可能是使用Airflow的Bash或Python运算符。你能直接从文件名中导出它吗?
如果是这样,那么您可以在GoogleCloudStorageObjectSensor上游使用Python运算符来检查文件是否在成功目录中。
如果是,则可以使用BigQueryOperator在BQ上运行删除查询。
之后,您运行GoogleCloudStorageToBigQueryOperator。
如果您要将文件从GCS移动到GCS位置,那么GoogleCloudStorageToGoogleCloudStorageOperator应该可以完成您需要的技巧。如果您的BQ加载操作员失败,则移至失败的文件位置,如果成功,则移至成功的作业位置。
也许您需要跟踪插入的所有内容都是将任务信息记录到GCS。看看how to log task information to GCS
这有帮助吗?