Как запустить потоковый запрос по обновленным строкам в CSV-файле?

Question

У меня есть один CSV-файл в папке, который постоянно обновляется.Мне нужно взять входные данные из этого CSV-файла и произвести некоторые транзакции.Как я могу взять данные из файла CSV, который постоянно обновляется, скажем, каждые 5 минут?

Я пытался со следующим:

val csvDF = spark
  .readStream
  .option("sep", ",")
  .schema(userSchema)
  .csv("file:///home/location/testFiles")

, но проблема в том, что он отслеживаетпапка, в которой были созданы какие-либо новые файлы или нет ... но моя проблема - только один файл, который постоянно обновляется.

Answer 1

У меня есть 1 CSV-файл в 1 папке, который постоянно обновляется.Мне нужно взять входные данные из этого CSV-файла и произвести некоторые транзакции.Как я могу взять данные из CSV-файла, который постоянно обновляется, скажем, каждые 5 минут.

tl; dr Это не будет работать.

Spark Structured Streaming по умолчанию отслеживает файлы в каталоге и для каждого нового файла запускает вычисления.Как только файл был обработан, файл никогда не будет обработан снова.Это реализация по умолчанию.

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

Answer 2

Во-первых, я не уверен, как вы сюда попали, потому что CSV-файл должен быть записан последовательно, что позволяет добиться лучшего ввода / вывода.Поэтому я рекомендую вам создать файл только для добавления и попытаться получить потоковые данные, такие как получение данных из binlog.

Однако, если вам нужно сделать это, я думаю, StreamingContext может вам помочь.

val ssc = new StreamingContext(new SparkConf(), Durations.milliseconds(1))
val fileStream = ssc.fileStream[LongWritable, Text, TextInputFormat]("/tmp", (x: Path) => true, newFilesOnly = false).map(_._2.toString)