数据流作业 - update_if_exists？答案

【问题标题】：Dataflow job - update_if_exists?数据流作业 - update_if_exists？
【发布时间】：2020-11-19 23:03:43
【问题描述】：

我使用 Google 数据流作业。我有以下命令来部署工作

java -jar build/libs/my-job-1.0-all.jar \
  --project=$PROJECT \
  --region=us-central1 \
  --subscription=$SUBSCRIPTION \
  --jobName=my-job \
  --runner=DataflowRunner \
  --streaming=true \
  --stableUniqueNames=ERROR \
  --workerMachineType=n1-standard-2 \
  --usePublicIps=false \
  --network=default \
  --update

当我在 GCP 中运行作业时，这可以正常工作，但如果没有要更新的内容，它会失败。如果我删除更新标志，只要没有正在运行的作业，它就可以正常工作。

有没有办法指定如果作业存在则更新它，而不仅仅是开始一个新作业？

【问题讨论】：

标签： google-cloud-dataflow apache-beam

【解决方案1】：

要更新您的工作，您需要启动一项新工作来替换正在进行的工作。当您启动替换作业时，除了作业的常规选项之外，您还必须设置以下管道选项来执行更新过程：

--update 选项
--jobName PipelineOptions 中的选项与您要更新的作业同名

如果您的管道中的任何转换名称已更改，您必须提供一个转换映射并使用--transformNameMapping 选项传递它。

请注意，--update 标志确保飞行中的数据不会丢失。

对于updating the job，在“启动替换作业”部分中指定了一种直接更新方法。此外，请注意：

目前不支持更新批处理管道。

有一点很重要，每次您要更新作业时，它必须在运行更新之前存在。如果要检查是否有正在进行的作业，可以使用 Dataflow 命令行界面准备一个脚本来查询现有作业的列表：gcloud beta dataflow jobs list，如果作业不存在则启动该作业，否则进行更新.

【讨论】：

我想我的问题并不清楚 - 我知道如何更新工作或如何启动新工作。我想知道是否有一种方法可以指定“upsert”，也就是说，如果没有正在进行的工作就开始新工作，如果有正在进行的工作进行更新。
目前还没有这样的选项，每次你想要更新作业时，它必须在运行更新之前存在。您可以做的是准备一个脚本，使用 Dataflow 命令行界面查询现有作业列表：gcloud beta dataflow jobs list，如果作业不存在则启动该作业，否则进行更新。