Spark结构化流媒体运行期间Presto提供的“不是拼花地板文件（太小）”

Question

我建立了一个管道，从Kafka读取数据，使用Spark结构化流处理数据，然后将拼花文件写入HDFS。数据查询的下游客户端正在使用配置为以配置单元表的形式读取数据的Presto。

Kafka--

一般来说，这是可行的。当Spark作业运行批处理时发生查询时，就会出现问题。Spark作业在HDFS上创建零长度拼花文件。如果Presto在处理查询的过程中试图打开此文件，则会抛出错误：

查询20171116_170937_07282_489cc失败：打开Hive拆分hdfs://namenode错误：50071/hive/仓库/表/part-00000-5a7c242a-3e53-46d0-9ee4-5d004ef4b1e4-c000.snappy.parquet（偏移量=0，长度=0）：hdfs://namenode：50071/hive/仓库/表/part-00000-5a7c242a-3e53-46d0-9ee4-5d004ef4b1e4-c000.snappy.parquet不是拼花文件（太小）

文件此时确实为零字节，因此错误严格正确，但这不是我希望管道的行为。我希望能够连续写入适当的HDFS文件夹，而不会干扰Presto查询。

作业的Spark scala代码如下所示：

val FilesOnDisk = 1
Spark
  .initKafkaStream("fleet_profile_test")
  .filter(_.name.contains(job.kafkaTag))
  .flatMap(job.parser)
  .coalesce(FilesOnDisk)
  .writeStream
  .trigger(ProcessingTime("1 hours"))
  .outputMode("append")
  .queryName(job.queryName)
  .format("parquet")
  .option("path", job.outputFilesPath)
  .start()

作业在小时的顶部：00开始。该文件在HDFS上首先在：05时作为零长度文件可见。直到在：21时完全写入它才会更新，就在作业完成之前。这使得表在Presto 25%的时间内实际上无法使用。

每个文件只有500kB多一点，所以我不希望文件的物理写入需要很长时间。据我所知，拼花文件的元数据位于文件的末尾，因此编写更大文件的人会遇到更多的麻烦。

在解决这个Presto错误的同时，人们使用了哪些策略来集成Spark结构化流媒体和Presto？

Answer 1

您可以尝试说服Presto（或Presto团队）忽略空文件，但这不会有帮助，因为编写文件的程序（此处为Spark）最终将刷新部分数据，并且文件将显示为部分、非空且格式不正确，因此也会导致错误。

防止Presto（或其他读取表数据的程序）看到部分文件的方法是将文件组装在不同的位置，然后原子地将文件移动到正确的位置。