数据流管道从 GCS 读取 csv 并通过调用 Vision 和 NL API 写入 BigBuery

Question

我想编写一个数据流程序（Java 和 maven 实现）。以下是我要执行的步骤：

1. Dataflow 应该从谷歌云存储中读取一个 csv 文件。 csv 文件格式如下：

   产品名称、图片网址、类别、描述1、描述2 Sakura 30062 6 件式 Pigma Micron 墨水笔套装，https://images-na.ssl-images-amazon.com/images/I/71CkvpG3FEL.SY355.jpg，Arts，包括 1 个尺寸：#005 (0.20mm)

   CCbetter 迷你热熔胶枪带 25 支胶棒高温熔胶枪套件 用于 DIY 小型工艺项目和密封和快速维修的灵活触发器（20 瓦，蓝色），https://images-na.ssl-images-amazon.com/images/I/61iFrMg4%2B3L.SY355.jpg，安全舒适的电源开关，带 LED 灯模式.具有可拆卸和灵活的支撑，使枪保持稳定和直立，高质量的绝缘喷嘴即使在 500°F 下长期使用也不会变形。

   。 . . .

2. 对于 csv 中的每一行，我需要选择图片 URL 并运行视觉 API 并获取前 2 个标签（例如，我们从视觉 API 为第一个产品/行以及 L3 和 L4 获取标签 L1 和 L2对于第二个产品/行）

3. 对于 csv 中的每一行，我需要连接产品名称、类别、描述 1 和描述 2，并将其传递给 NL API。从 NL API 的响应中，我需要选择消费品类别下的前 2 个实体（例如，我们从第一行获得 E1 和 E2，从第二行获得 E3 和 E4）

4. 我需要从检索到的响应中创建以下结构：

   产品名称、主题 Sakura 30062 6 支 Pigma Micron 墨水笔套装，L1 Sakura 30062 6 支 Pigma Micron 墨水笔套装，L2 Sakura 30062 6 支 Pigma Micron 墨水笔套装，E1 Sakura 30062 6 支 Pigma Micron 墨水笔套装，E2

   CCbetter 迷你热熔胶枪带 25 根胶棒高温熔胶枪套件 用于 DIY 小型工艺项目和密封和快速维修的灵活触发器（20 瓦，蓝色），L3 CCbetter 迷你热熔胶枪带 25 根胶棒高温熔胶枪套件 用于 DIY 小型工艺项目和密封和快速维修的灵活触发器（20 瓦，蓝色），L4 CCbetter 迷你热熔胶枪带 25 支胶棒高温熔胶枪套件 用于 DIY 小型工艺项目和密封和快速维修的灵活触发器（20 瓦，蓝色），E3 CCbetter 迷你热熔胶枪带 25 根胶棒高温熔胶枪套件 用于 DIY 小型工艺项目和密封和快速维修的灵活触发器（20 瓦，蓝色），E4 . . . .

5. 我想将此网格（步骤 4 中的结构）写入 Bigquery 表

我是 Dataflow 的新手，因此非常感谢任何帮助、代码 sn-p 或整个源代码或参考

Answer 1

您应该首先阅读quick start guides 中的一个，然后看看example pipelines 中的一些。

根据您的描述，高级大纲可能是：

1. 使用TextIO.read 从 GCS 读取内容。请注意，它不支持忽略标头，因此您可能需要自己检测并删除它。
2. 在文件每一行的 URL 上编写一个使用视觉 API 的 DoFn。您甚至可以将其分成多个 DoFns - 一个用于将行转换为 URL，然后是一个 DoFn 以使用视觉 API，然后是一个 DoFn 以提取前两个标签。
3. 编写另一个 DoFn 或一系列 DoFns 来执行连接并使用 NL API。
4. 编写另一个DoFn 或一系列DoFns，生成具有所需输出格式的行TableRows。
5. 使用 BigQueryIO.write 转换将这些内容写入 BigQuery。