我正在从消息传递应用程序收集数据,我目前正在使用 Flume,它每天发送大约 5000 万条记录
我想用卡夫卡,
使用 Spark Streaming 从 Kafka 消费
并将其保存到 hadoop 并使用 impala 进行查询
我尝试过的每种方法都遇到问题..
方法 1 - 将 RDD 保存为 parquet,将外部 hive 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。另存为Parquet文件输出大量文件,从 Kafka 接收的 Dstream 在 1 分钟批量大小下输出超过 200 个文件。
它输出许多文件的原因是因为计算是分布式的,如另一篇文章中所解释的那样如何使 saveAsTextFile 不将输出拆分为多个文件? https://stackoverflow.com/questions/24371259/how-to-make-saveastextfile-not-split-output-into-multiple-file/24378808#24378808
然而,所提出的解决方案对我来说似乎并不是最佳的,例如正如一位用户所说 - 如果您的数据很少,那么只有一个输出文件才是一个好主意。
方法 2 - 使用HiveContext
。直接将RDD数据插入到Hive表中
# 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);
这工作正常,它直接插入到镶木地板表中,但由于处理时间超过批处理间隔时间,因此批处理会出现调度延迟。
消费者无法跟上正在生产的产品,并且要处理的批次开始排队。
看来写入 hive 很慢。我尝试调整批处理间隔大小,运行更多消费者实例。
总之
考虑到存在多个文件的问题以及写入 hive 的潜在延迟,保存 Spark Streaming 中的大数据的最佳方法是什么?
其他人在做什么?
这里已经提出了类似的问题,但他对目录有一个问题,而不是太多文件如何让 Spark Streaming 写入其输出以便 Impala 可以读取它? https://stackoverflow.com/questions/24204656/how-to-make-spark-streaming-write-its-output-so-that-impala-can-read-it
非常感谢您的帮助