Как обновить существующую запись в потоковом приемнике ORC?

Question

При сохранении потока в файл, используя формат файла Apache ORC, есть ли способ выполнить обновление записи?Вместо добавления и эффективного наличия записи несколько раз при обновлении записи.

incomingStreamDF.writeStream
  .format("orc")
  .option("path", "/mnt/adls/orc")
  .option("checkpointLocation", "/mnt/adls/orc/check")
  .trigger(ProcessingTime("25 seconds"))
  .start()

Похоже, что ORC поддерживает обновление, поэтому есть способ указать ключ записи, возможно, в параметрах writeStream.

Answer 1

tl; dr Нет (до Spark 2.4 включительно)

Единственный выходной режим, который может предоставить вам такую ​​функцию, будет Update режим вывода.Поскольку orc формат FileFormat, он * всегда должен использоваться с режимом добавления вывода .

---

Решением проблемы может быть использование совершенно нового Оператор DataStreamWriter.foreachBatch (или более старый DataStreamWriter.foreach ), где вы обрабатываете данные так, как вам нравится (и вы можете легко обновить запись в файле ORC, если знаете, как это сделать).

foreachBatch (функция: (набор данных [T], длинный) ⇒ единица измерения): DataStreamWriter [T]

Устанавливает вывод потокового запроса равнымобрабатывается с использованием предоставленного function.

Это поддерживается только в режимах микропакета (то есть, когда триггер не является непрерывным).

Предоставленная функция будет вызываться вкаждый микропакет с:

(i) выходными строками в виде набора данных

(ii) идентификатором пакета.

batchId может использоваться для дедупликации и записи транзакцийвывод (то есть предоставленный набор данных) во внешнюю системуems.

Выходной набор данных гарантированно совпадает для одного и того же batchId (при условии, что все операции являются детерминированными в запросе).