как использовать сообщение кафки не с самого начала после неудачной работы с искрой

Question

Я новичок в kafka и spark, удивляюсь, как восстановить смещение от kafka после сбоя задания spark.

Условия:

1. скажем, 5 Гбит / с потока Кафки, с самого начала трудно потреблять

2. потоковые данные уже были использованы, поэтому как сказать, чтобы spark повторно использовал сообщение / плавно повторил неудачную задачу

Я не уверен, какую область искать, возможно, кто-то может указать мне правильное направление

Answer 1

Когда мы имеем дело с кафкой, у нас должно быть 2 разные темы. Один за успех, другой за неудачу.

Допустим, у меня есть 2 темы Тема-Успех и Тема-Сбой. Когда Kafka успешно обрабатывает поток данных, мы можем пометить его и сохранить в теме Topic-Success, а когда Kafka не сможет обработать поток данных, тогда он сохранит его в теме с ошибкой Topic.

Так что, когда вы захотите повторно использовать поток данных с ошибками, мы сможем обработать этот поток из темы с ошибкой темы. Здесь вы можете исключить повторное использование всех данных с начала.

Надеюсь, это поможет вам.

Answer 2

В kafka 0.10.x есть концепция Consumer Group, которая используется для отслеживания смещения сообщений.

Если вы установили enable.auto.commit = true и auto.offset.reset = latest, он не будет потребляться с начала. Теперь, используя этот подход, вам также может понадобиться отслеживать смещения, поскольку процесс может завершиться неудачей после потребления. Я бы посоветовал вам использовать этот метод, предложенный в Spark Docs для

stream.foreachRDD { rdd =>
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

  // some time later, after outputs have completed
  stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
} 

CanCommitOffsets находится в ваших руках, чтобы передать эти сообщения, когда ваш сквозной конвейер будет оправдан