自定义无界源如何在Google Cloud DataFlow中工作？

Question

我正在尝试将自定义无界源实现到Google Cloud Dataflow以从Amazon Kinesis队列中读取。为了正确实施检查点，我想了解该机制的确切工作原理。

我试图通过阅读数据流文档来理解检查点，但缺少了一些关键的东西，所以我阅读了一篇磨盘论文。首先让我解释一下我是如何理解本文中提出的概念的。在数据流API方面，我将重点关注在强大的生产设置中源和其消费者之间的交互：

• 在源上调用createReader（），并将null值作为检查点标记传递

现在，请允许我对Kinesis队列的操作方式说几句话，因为它与Pub/Sub有很大的不同（因为我了解Pub/Sub的工作原理，但我自己并没有使用它）。

我看到发布/订阅拉式模型严重依赖于ACK，即客户端接收的消息被ACK，然后发布/订阅中的“内部检查点”向前移动-

动觉拉界面（这里根本没有推）更类似于您与文件交互的方式。您可以在流中的任何位置开始读取（特殊值TRIM\u HORIZON是流中最旧的记录，LATEST是流中最新的记录），然后使用迭代器逐个记录向前移动（迭代器存储在服务器端，如果未使用，则有5分钟的到期时间）。服务器没有ACK—客户机负责跟踪流中的位置，您可以随时重新读取旧记录（当然，除非它们已过期）。

• 检查点应该是什么样子？给定检查点的读卡器是否只需要读取与之相关的部分数据，还是从检查点读取所有数据？换句话说，我的检查点应该是“x和y之间的数据”还是“x之后的所有数据”
• 我知道第一个读卡器作为检查点标记得到null，这很好——这意味着我应该从应用程序开发人员定义的点开始读取。但DataFlow是否可以创建其他具有null的读取器（例如，我可以想象这样的情况：读取器jvm死亡，然后DataFlow创建一个新的读取器，新读取器将null作为检查点传递）？在这种情况下，我不知道我的起始位置是什么，因为我可能已经使用以前的阅读器阅读了一些数据，现在进度的标记已经丢失
• 消费者端记录重复数据消除使用的id是什么？它是由getCurrentRecordId返回的值吗？我问这个问题，因为我考虑过使用流中的位置，因为它对于特定流是唯一的。但如果我后来通过扁平化加入少数动觉源，会发生什么-

干杯，Przemek

Answer 1

我们很高兴看到您将数据流与运动相结合。我们希望通过contrib连接器为GitHub项目提供pull请求。我们也很乐意在您开发时通过GitHub查看您的代码，并在那里向您提供反馈。

检查点应该是什么样子？给定检查点的读卡器是否只需要读取与之相关的部分数据，还是从检查点读取所有数据？换句话说，我的检查点应该是“x和y之间的数据”还是“x之后的所有数据”？

检查点标记应表示“该读卡器已生成并最终确定的数据”。E、 例如，如果一个读卡器负责一个特定的分片，检查点标记可能由分片标识符和该分片中已成功读取的最后一个序列号Y组成，表示“已生成Y之前和包括Y在内的所有数据”。

我知道第一个读卡器作为检查点标记得到null，这很好——这意味着我应该从应用程序开发人员定义的点开始读取。但DataFlow是否可以创建其他具有null的读取器（例如，我可以想象这样的情况：读取器jvm死亡，然后DataFlow创建一个新的读取器，新读取器将null作为检查点传递）？在这种情况下，我不知道我的起始位置是什么，因为我可能已经使用以前的阅读器阅读了一些数据，现在进度的标记已经丢失。

即使在JVM发生故障时，最终确定的检查点也会保持不变。换句话说，当JVM死亡时，读取器将使用最后确定的检查点来构造。您不应该看到使用空检查点创建的读卡器，除非它们打算从源代码的开头读取，或者在您的场景中，当JVM在第一次成功调用finalizeCheckpoint（）之前死亡时。您可以在新的读卡器上使用检查点标记，为从要读取的下一条记录开始的同一个碎片构造一个新的迭代器，并且可以在不丢失数据的情况下继续。

消费者端记录重复数据消除使用的id是什么？它是getCurrentRecordId返回的值吗？我问这个问题，因为我考虑过使用流中的位置，因为它对于特定流是唯一的。但如果我后来通过扁平化加入少数动觉源，会发生什么-

在数据流中，每个无界源（实现getCurrentRecordId并覆盖RequiresDuplication以返回true）都会自行进行重复数据消除。因此，对于同一个源实例，记录ID只需要是唯一的。来自两个不同来源的记录可以使用相同的记录ID，在展平过程中，它们不会被视为“重复”。因此，如果Amazon Kinesis保证所有记录的ID都是全局唯一的（跨流中的所有碎片）和持久的（例如跨重排操作），那么这些ID应该适合用作记录ID。

请注意，getCurrentRecordId是无界恐惧者的可选方法，如果检查点方案唯一标识每个记录，则无需实现它。Kinesis允许您按序号顺序读取记录，看起来序号是全局唯一的。因此，您可以在generateInitialSplits中将每个shard分配给不同的worker，并且每个worker可能永远不会产生重复的数据——在这种情况下，您可能根本不需要担心记录ID。

大多数答案都假设了一个简单的情况，即你的运动流永远不会改变它们的碎片。另一方面，如果流上的切分发生变化，那么您的解决方案将变得更加复杂。E、 例如，每个工作人员可以负责1个以上的碎片，因此检查点标记将是碎片地图-