使用 RabbitMQ 源的 Spark 结构化流

I want my reciever to be reliable, i.e. after every processed message (or at least written to chekpoint directory for further processing), I need to ack it back to RabbitMQ. The problem starts after here: these factories are created at the driver, and the actual reading process takes place at executors through DataReaders. However, the commit method is a part of MicroBatchReader, not DataReader. Since I have many DataReaders per MicroBatchReader, how should I ack these messages back to RabbitMQ? Or should I ack when the next method is called on DataReader? Is it safe? If so, what is the purpose of commit function then?

澄清: OBFUSCATION:答案中提供的有关重命名某些类/函数的链接(除了那里的解释)让一切更加清晰 比以往任何时候都更糟.引用 那里:

CLARIFICATION: OBFUSCATION: The link provided in the answer about the renaming of some classes/functions (in addition to the explanations there) made everything much more clear worse than ever. Quoting from there:

重命名:

• DataReaderFactory 到 InputPartition

DataReader 到 InputPartitionReader

...

InputPartition 的目的是管理关联的阅读器，现在称为 InputPartitionReader，带有显式创建操作以镜像关闭操作.这是没有从 API 中清除的时间更长，因为 DataReaderFactory 似乎更多比它更通用，并且不清楚为什么要生产一组它们阅读.

InputPartition's purpose is to manage the lifecycle of the associated reader, which is now called InputPartitionReader, with an explicit create operation to mirror the close operation. This was no longer clear from the API because DataReaderFactory appeared to be more generic than it is and it isn't clear why a set of them is produced for a read.

但是，docs 明确表示读取器工厂将被序列化并发送到执行器，然后将在执行器上创建数据读取器并进行实际读取."

However, the docs clearly say that "the reader factory will be serialized and sent to executors, then the data reader will be created on executors and do the actual reading."

为了使消费者可靠，我必须仅在特定消息在 Spark 端提交后才对它进行 ACK.请注意消息必须在传递消息的同一连接上进行确认，但在驱动程序节点调用提交函数.如何在 worker/executor 节点上提交?

To make the consumer reliable, I have to ACK for a particular message only after it is committed at Spark side. Note that the messages have to be ACKed on the same connection that it has been delivered through, but commit function is called at driver node. How can I commit at the worker/executor node?

推荐答案

> 我正在返回一个工厂列表，并希望列表中的每个实例都用于创建一个读取器，它也是一个消费者.这种方法正确吗?源 [socket][1] 源实现有一个线程将消息推送到内部 ListBuffer.换句话说，有一个消费者(线程)填充了内部 ListBuffer，它**然后**被`planInputPartitions`划分为多个分区(`createDataReaderFactories` [renamed][2] 到 `planInputPartitions`).此外，根据 [MicroBatchReadSupport][3] 的 Javadoc> 执行引擎将在流式查询开始时创建一个微批处理读取器，为每个要处理的批处理交替调用 setOffsetRange 和 createDataReaderFactories，然后在执行完成时调用 stop().请注意，由于重新启动或故障恢复，单个查询可能会执行多次.换句话说，`createDataReaderFactories` 应该被调用 **multiple** 次，据我了解，这表明每个 `DataReader` 负责一个静态输入分区，这意味着 DataReader 不应该是消费者.----------> 但是，commit 方法是 MicroBatchReader 的一部分，而不是 DataReader ... 如果是这样，那么 commit 函数的目的是什么?提交函数的部分基本原理可能是防止 MicroBatchReader 的内部缓冲区变大.通过提交偏移量，您可以有效地从缓冲区中删除小于偏移量的元素，因为您承诺不再处理它们.您可以使用 `batches.trimStart(offsetDiff)` 在套接字源代码中看到这种情况<小时><删除>我不确定是否要实现一个可靠的接收器，所以我希望一个更有经验的 Spark 人能过来解决你的问题，因为我也有兴趣！希望这可以帮助！

编辑

我只研究了 socket 和 wiki-edit 来源.这些资源还没有准备好生产，这是问题所在.相反， kafka 源是更好的起点，与前面提到的源不同，它有多个像作者一样的消费者正在寻找.

I had only studied the socket, and wiki-edit sources. These sources are not production ready, which is something that the question was was not looking for. Instead, the kafka source is the better starting point which has, unlike the aforementioned sources, multiple consumers like the author was looking for.

但是，也许如果您正在寻找不可靠的来源，上面的套接字和 wikiedit 来源提供了一个不太复杂的解决方案.

However, perhaps if you're looking for unreliable sources, the socket and wikiedit sources above provide a less complicated solution.

这篇关于使用 RabbitMQ 源的 Spark 结构化流的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持跟版网！