问题标签 [google-cloud-dataflow]

我创建了一个管道，它从 GCS 中的文件读取，对其进行转换，最后写入 BQ 表。该文件包含一个标题行（字段）。

有没有办法像在加载时在 BQ 中那样以编程方式设置“要跳过的标题行数”？

我正在尝试在 Dataflow 中执行联合操作。是否有在 Dataflow 中合并两个 PCollection 的示例代码？

阅读 Google 的 Dataflow API，我的印象是它与 Apache Storm 所做的非常相似。通过流水线流进行实时数据处理。除非我完全错过了这里的重点，而不是建立关于如何执行相互编写的管道的桥梁，我希望有一些与谷歌不同的东西，而不是重新发明轮子。Apache Storm 已经很好地适应了任何编程语言。做这样的事情的真正价值是什么？

在管道中，是否可以在另一个云项目中从/向云存储文件执行 TextIO？

使用“my-project:output.output_table”并正确设置服务帐户似乎可以访问另一个项目中的 BigQuery 表。

但是，使用 TextIO，我无法找到一种方法来指定项目 ID 以及我的文件模式“gs://some/inputData.txt”。

寻找一些关于如何最好地架构/设计和构建我们的管道的建议。

经过一些初步测试，我们没有得到预期的结果。也许我们只是在做一些愚蠢的事情，或者我们的期望太高了。

我们的数据/工作流程：

• Google DFP 将我们的广告服务器日志（CSV 压缩）直接写入 GCS（每小时）。
• 这些日志一天的价值在 30-7000 万条记录范围内，一个月大约有 1.5-20 亿条记录。
• 对其中的 2 个字段执行转换，并将行写入 BigQuery。
• 转换涉及对其中 2 个字段执行 3 次 REGEX 操作（由于增加到 50 个操作），这会产生新的字段/列。

到目前为止，我们已经运行了什么：

• 构建了一个从 GCS 读取文件一天（31.3m）的管道，并使用 ParDo 执行转换（我们认为我们会从一天开始，但我们的要求也是处理数月和数年）。
• DoFn 输入是一个字符串，它的输出是一个 BigQuery TableRow。
• 管道在云中以实例类型“n1-standard-1”（1vCPU）执行，因为我们认为每个工作人员 1 个 vCPU 就足够了，因为转换不是过于复杂，也不是 CPU 密集型，即只是字符串到字符串的映射.

我们使用几种不同的工作器配置运行了该作业，以查看它的执行情况：

1. 5 名工作人员（5 个 vCPU）耗时约 17 分钟
2. 5 个工作人员（10 个 vCPU）花费了大约 16 分钟（在这次运行中，我们将实例提升到“n1-standard-2”以获得双倍的内核，看看它是否提高了性能）
3. 自动缩放设置为“BASIC”（50-100 个 vCPU）的 50 分钟和最多 100 个工作人员花费了大约 13 分钟
4. 自动缩放设置为“BASIC”（100-150 个 vCPU）的 100 分钟和 150 个最大工作人员花费了大约 14 分钟

这些时间是否符合您对我们的用例和管道的期望？

我知道可以在使用时指定管道输入TextIO.Read的压缩类型，如下所示：

是否也可以在使用时指定管道输出TextIO的压缩类型？我在 API 中看不到任何关于压缩的内容TextIO.Read.from(..)。

在云中执行我们的管道运行良好。但是当它作为一个DirectPipelineRunner（即本地）运行它时，它会出错，并抱怨提供的文件模式。文件模式使用 glob。

这是在本地运行时的预期行为吗？

是否可以使用运行器（异步）提交 X 个单独的作业/管道DataflowPipelineRunner，并让这些作业共享工作池，而不是启动并拆除单个作业？

我们的工作/管道正在将 ParDo 转换的结果写回 GCS，即使用TextIO.Write.to("gs://...")

我们注意到，当作业/管道完成时，它会在输出桶中留下许多 0 字节的文件。

管道的输入来自 GCS 的多个文件，所以我假设结果是分片的，这很好。

但是为什么我们会得到空文件呢？

我们创建了一个管道，它正在对位于 GCS 中的 3 个流（“Clicks”、“Impressions”、“ActiveViews”）执行转换。我们需要将各个流写回 GCS，但要分开文件（稍后加载到 BigQuery），因为它们都有稍微不同的架构。

其中一个写入连续两次失败，每次都有不同的错误，这反过来导致管道失败。

这些是 GDC 以可视方式表示的最后 2 个工作流/管道，它们显示了失败：

第一个错误：

第二个错误：

它只发生在“ActiveViews-GCS-Write”步骤中。

知道我们做错了什么吗？