Как получить смещения Kafka для структурированного запроса для ручного и надежного управления смещением?

Ответ 1

Актуальная дискуссионная дискуссионная тема Spark DEV здесь.

Резюме из него:

Spark Streaming будет поддерживать смещение в будущих версиях ( > 2.2.0). JIRA билет - https://issues-test.apache.org/jira/browse/SPARK-18258

Для Spark & ​​lt; = 2.2.0 вы можете получить смещения для данной партии, прочитав json из каталога контрольной точки (API нестабилен, поэтому будьте осторожны):

val checkpointRoot = // read 'checkpointLocation' from custom sink params
val checkpointDir = new Path(new Path(checkpointRoot), "offsets").toUri.toString
val offsetSeqLog = new OffsetSeqLog(sparkSession, checkpointDir)

val endOffset: Map[TopicPartition, Long] = offsetSeqLog.get(batchId).map { endOffset =>
  endOffset.offsets.filter(_.isDefined).map { str =>
    JsonUtilsWrapper.jsonToOffsets(str.get.json)
  }
}


/**
  * Hack to access private API
  * Put this class into org.apache.spark.sql.kafka010 package
  */
object JsonUtilsWrapper {
  def offsetsToJson(partitionOffsets: Map[TopicPartition, Long]): String = {
    JsonUtils.partitionOffsets(partitionOffsets)
  }

  def jsonToOffsets(str: String): Map[TopicPartition, Long] = {
    JsonUtils.partitionOffsets(str)
  }
}

Этот endOffset будет содержать до смещения для каждого раздела/раздела. Получение стартовых смещений является проблематичным, потому что вам нужно прочитать "контрольный" контрольный адрес. Но, как правило, вы не заботитесь о стартовых смещениях, потому что хранения окончательных смещений достаточно для надежного возобновления работы Spark.

Обратите внимание, что вы также должны хранить обработанный пакетный идентификатор в своем хранилище. В некоторых случаях Spark может повторно запускать неудачную партию с одним и тем же идентификатором пакета, поэтому обязательно инициализируйте пользовательскую раковину с последним обработанным идентификатором пакета (который вы должны прочитать из внешнего хранилища) и проигнорируйте любую партию с id < latestProcessedBatchId. Btw, пакетный идентификатор не является уникальным для всех запросов, поэтому вам нужно хранить пакетный идентификатор для каждого запроса отдельно.

Ответ 2

Spark 2.2 представил структурированный потоковый источник Kafka. Насколько я понимаю, он полагается на контрольную точку HDFS для хранения смещений и гарантирует доставку сообщений "точно один раз".

Правильно.

Каждый триггер Spark Structured Streaming будет сохранять смещения в каталог offset в месте контрольной точки (определяется с помощью опции checkpointLocation или spark.sql.streaming.checkpointLocation свойство Spark или произвольно назначается), который должен гарантировать, что смещения обрабатываются в чаще всего. Эта функция называется Write Ahead Logs.

Другим каталогом в местоположении контрольной точки является каталог commits для завершенных пакетов потоковой передачи с одним файлом за пакет (с именем файла, являющимся идентификатором пакета).

Указание официальной документации в Семантика допусков отказов:

Для этого мы разработали источники структурированных потоков, приемники и механизм выполнения, чтобы надежно отслеживать точный ход обработки, чтобы он мог справляться с любым отказом путем перезапуска и/или переработки. Предполагается, что каждый источник потоковой передачи имеет смещения (аналогичные смещениям Kafka или номерам последовательностей Kinesis) для отслеживания позиции чтения в потоке. Для записи диапазона смещения данных, обрабатываемых в каждом триггере, двигатель использует контрольные и записываемые журналы. Потоковые раковины предназначены для идемпотента для обработки переработки. Вместе с использованием воспроизводимых источников и идемпотентных поглотителей Structured Streaming может обеспечить сквозную семантику точно один раз при любом сбое.

Каждый раз, когда запускается триггер StreamExecution проверяет каталоги и "вычисляет", какие смещения уже обработаны. Это дает вам хотя бы раз семантику и ровно один раз.

Но старые документы (...) говорят, что контрольные точки Spark Streaming не восстанавливаются в приложениях или обновлениях Spark и, следовательно, не очень надежны.

Была причина, почему вы назвали их "старыми", не было?

Они относятся к старой и (по моему мнению) мертвой Spark Streaming, которая хранит не только смещения, но и весь код запроса, который привел к ситуациям, когда контрольная точка была почти непригодна для использования, например. при изменении кода.

Время закончилось, и Structured Streaming более осторожно, что и когда проверяется.

Если я хочу хранить смещения от источника Кафки до транзакционной БД, как я могу получить смещение от партии структурированного потока?

Решением может быть реализация или каким-то образом использование MetadataLog интерфейса, который используется для обработки контрольной точки смещения. Это может сработать.

Как я могу предположить, чтобы получить смещение для заданного идентификатора пакета?

В настоящее время это невозможно.

Я понимаю, что вы не сможете это сделать, поскольку семантика потоковой передачи скрыта от вас. Вы просто должны не иметь дело с этой низкоуровневой "вещью", называемой смещениями, которую Spark Structured Streaming использует, чтобы предложить ровно один раз гарантии.

Цитата Майкла Армбруста из его выступления на Spark Summit Простая, масштабируемая, отказоустойчивая обработка потока со структурированным потоком в Apache Spark:

вам не нужно будет рассуждать о потоковой передаче

и далее в разговоре (на следующем слайде):

вы должны написать простые запросы, и Spark должен постоянно обновлять ответ

Там есть способ получить смещения (из любого источника, включая Kafka), используя StreamingQueryProgress, который вы можете перехватить с помощью StreamingQueryListener и onQueryProgress.

onQueryProgress (событие: QueryProgressEvent): Unit Вызывается при наличии некоторого обновления статуса (скорость приема пищи и т.д.)

С помощью StreamingQueryProgress вы можете получить доступ к свойству sources с помощью SourceProgress, который дает вам то, что вы хотите.

Ответ 3

Потоковый набор данных с источником Кафки имеет offset как один из полей . Вы можете просто запросить все смещения в запросе и сохранить их в JDBC Sink