已经实现了以下目标
Kafka制作人使用spark流媒体从twitter上获取数据。
kafka使用者将数据摄取到hive外部表（在hdfs上）。
虽然目前为止效果还不错。我只面临一个问题，当我的应用程序将数据插入配置单元表时，它创建了一个小文件，每个文件的每一行都有数据。
下面是代码
// Define which topics to read from
val topic = "topic_twitter"
val groupId = "group-1"
val consumer = KafkaConsumer(topic, groupId, "localhost:2181")
//Create SparkContext
val sparkContext = new SparkContext("local[2]", "KafkaConsumer")
//Create HiveContext
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sparkContext)
hiveContext.sql("CREATE EXTERNAL TABLE IF NOT EXISTS twitter_data (tweetId BIGINT, tweetText STRING, userName STRING, tweetTimeStamp STRING, userLang STRING)")
hiveContext.sql("CREATE EXTERNAL TABLE IF NOT EXISTS demo (foo STRING)")配置单元演示表已经填充了一条记录。kafka consumer循环处理每一行中topic=“topic\u twitter”的数据并填充到配置单元表中
val hiveSql = "INSERT INTO TABLE twitter_data SELECT STACK( 1," +
tweetID +"," +
tweetText +"," +
userName +"," +
tweetTimeStamp +"," +
userLang + ") FROM demo limit 1"
hiveContext.sql(hiveSql)下面是来自我的hadoop环境的图片。推特数据，演示
在hdfs中创建的最后10个文件
如您所见，文件大小不超过200kb，有没有办法将这些文件合并到一个文件中？
3条答案
按热度按时间ws51t4hk1#
[take 2]好的，所以您不能将数据正确地“流式传输”到hive中。但是你可以添加一个周期性的压缩后处理作业。。。
用3个分区创建表。
(role='collectA'),(role='collectB'),(role='archive')把你的Spark塞指向(role='activeA')在某个时刻，切换到(role='activeB')然后将您在“a”分区中收集的所有记录转储到“archive”中，希望配置单元默认配置能够很好地限制碎片INSERT INTO TABLE twitter_data PARTITION (role='archive') SELECT ... FROM twitter_data WHERE role='activeA' ; TRUNCATE TABLE twitter_data PARTITION (role='activeA') ;在某个时刻，切换回“a”等。最后一句话：如果配置单元仍然在每个压缩作业上创建太多的文件，那么在插入之前尝试调整会话中的一些参数。
qybjjes12#
hive是为大规模批量处理而设计的，而不是为事务而设计的。这就是为什么每个load或insert select命令至少有一个数据文件。这也是为什么没有insert values命令的原因，因此本文中显示的蹩脚语法是一种必要的解决方法。
好。。。在引入事务支持之前，情况一直如此。简而言之，您需要（a）hivev0.14和更高版本（b）orc表（c）在该表上启用事务支持（即锁、定期后台压缩等）
关于在hive中接收流数据的wiki可能是一个好的开始。
yhqotfr83#
您可以同时使用这些选项。
打开酸
创建具有事务属性的orc表k。
在k中插入多次。通过流式传输或仅使用insert dml。
配置单元将自动创建小增量文件
小的或大的压缩会发生
小文件将合并为大文件。