独立的 Kafka Spark Sinks（多个生产者和经纪人）

Question

所以我在将 JSON 发送到多个主题和 不可靠的 kafka 代理时遇到了 Spark Streaming 中的 Kafka 接收器问题。以下是部分代码：

val kS = KafkaUtils.createDirectStream[String, TMapRecord]
(ssc,
PreferConsistent,
Subscribe[String, TMapRecord](topicsSetT, kafkaParamsInT))

然后我遍历 RDD 的

kSMapped.foreachRDD {
  rdd: RDD[TMsg] => {
    rdd.foreachPartition {
      part => {
        part.foreach { ........... 

我在 foreach 里面做

kafkaSink.value.send(kafkaTopic, strJSON)

kafkaSinkMirror.value.send(kafkaTopicMirrorBroker, strJSON)

当镜像代理关闭时，整个流应用程序都在等待它，我们不会向主代理发送任何内容。

你会怎么处理？

对于您提出的最简单的解决方案，假设我只是跳过本应发送给发生故障的代理的消息（例如，情况 1）

对于 CASE 2，我们会做一些缓冲。

附：稍后我将使用 Kafka Mirror，但目前我没有这样的选项，所以我需要在我的代码中做一些解决方案。

Answer 1

我发现了这个问题的几个决定：

1. 您可以使用在 worker 和 checkpoints 上抛出任何超时异常。 Spark 尝试多次重新启动错误任务，在 spark.task.maxFailures 属性中进行了描述。可以增加重试次数。如果流式作业在最大重试后失败，只会在代理可用时从检查点重新启动作业。或者你可以manually stop the job when it fails。
2. 您可以配置 backpressure spark.streaming.backpressure.enabled=true 以允许在处理数据时以最快的速度接收数据。
3. 您可以将两个结果发送回您的技术 Kafka 主题，稍后通过另一个流作业处理。
4. 您可以为这种情况制作 Hive 或 Hbase 缓冲区，稍后以批处理模式发送未处理的数据。