我正在从一个消息应用程序收集数据，我目前正在使用flume，它每天发送大约5000万条记录
我希望使用kafka，使用spark streaming从kafka消费，并将其持久化到hadoop和impala查询
我尝试过的每种方法都有问题。。
方法1-将rdd另存为parquet，将外部配置单元parquet表指向parquet目录

// scala
val ssc =  new StreamingContext(sparkConf, Seconds(bucketsize.toInt))
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
lines.foreachRDD(rdd => {

    // 1 - Create a SchemaRDD object from the rdd and specify the schema
    val SchemaRDD1 = sqlContext.jsonRDD(rdd, schema)

    // 2 - register it as a spark sql table
    SchemaRDD1.registerTempTable("sparktable")

    // 3 - qry sparktable to produce another SchemaRDD object of the data needed 'finalParquet'. and persist this as parquet files
    val finalParquet = sqlContext.sql(sql)
    finalParquet.saveAsParquetFile(dir)

问题是finalparquet.saveasparquetfile输出大量文件，从kafka接收的数据流以1分钟的批量大小输出200多个文件。它输出许多文件的原因是因为计算是分布式的，正如另一篇文章中所解释的-如何使saveastextfile不将输出拆分为多个文件？
然而，建议的解决方案似乎并不适合我，例如，正如一个用户所说，只有一个输出文件是一个好主意，如果你有很少的数据。
方法2-使用 HiveContext . 将rdd数据直接插入配置单元表

# python

sqlContext = HiveContext(sc)
ssc = StreamingContext(sc, int(batch_interval))
kvs = KafkaUtils.createStream(ssc, zkQuorum, group, {topics: 1})
lines = kvs.map(lambda x: x[1]).persist(StorageLevel.MEMORY_AND_DISK_SER)
lines.foreachRDD(sendRecord)

def sendRecord(rdd):

  sql = "INSERT INTO TABLE table select * from beacon_sparktable"

  # 1 - Apply the schema to the RDD creating a data frame 'beaconDF'
  beaconDF = sqlContext.jsonRDD(rdd,schema)

  # 2- Register the DataFrame as a spark sql table.
  beaconDF.registerTempTable("beacon_sparktable")

  # 3 - insert to hive directly from a qry on the spark sql table
  sqlContext.sql(sql);

这工作得很好，它直接插入到Parquet地板表中，但是由于处理时间超过了批间隔时间，因此批的调度会出现延迟。消费者跟不上正在生产的产品，要加工的批次开始排队。
向Hive写信似乎很慢。我尝试过调整批处理间隔大小，运行更多的使用者示例。

总而言之

如果存在多个文件的问题以及写入配置单元的潜在延迟，那么保存spark流媒体中的大数据的最佳方法是什么？其他人在干什么？
这里也有人问过类似的问题，但他有一个问题，目录中有太多的文件，如何让spark流写入其输出，以便 Impala 可以读取它？
非常感谢你的帮助