Hive为HDFS中的每个插入创建多个小文件

Question

以下是已经实现的

1. Kafka制作人使用Spark流媒体从推特上获取数据
2. 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”的数据，并填充到Hive表中

val hiveSql = "INSERT INTO TABLE twitter_data SELECT STACK( 1," + 
    tweetID        +","  + 
    tweetText      +"," + 
    userName       +"," +
    tweetTimeStamp +","  +
    userLang + ") FROM demo limit 1"

hiveContext.sql(hiveSql)

如您所见，文件大小不超过200KB，有没有办法将这些文件合并到一个文件中？

Answer 1

您可以同时使用这些选项。

1. 打开acid

Answer 2

Hive是为大规模批次处理作业而设计的，而不是为事务而设计的。这就是为什么每个LOAD或INSERT-SELECT命令至少有一个数据文件。这也是为什么您没有INSERT-VALUES命令，因此您的帖子中显示的蹩脚语法是必要的解决方法。

好在引入事务支持之前，情况一直如此。简而言之，您需要（a）Hive V0.14和更高版本（b）ORC表（c）在该表上启用事务支持（即锁、定期后台压缩等）

关于Hive中流式数据摄取的wiki可能是一个好的开始。

Answer 3

[采取2]好吧，所以你不能正确地将数据“流式传输”到Hive。但是你可以添加一个周期性的压缩后处理作业...

>

使用3个分区创建您的表，例如（角色='集合A'）、（角色='集合B'）、（角色='存档'）

将您的Spark插入指向（角色='activeA'）

在某个时候，切换到（角色='activeB'）

然后将您在“A”分区中收集的每条记录转储到“存档”中，希望Hive默认配置能很好地限制碎片

INSERT INTO TABLE twitter\u data PARTITION（role='archive'）选择。。。来自twitter\u数据，其中role='activeA'；截断表twitter\u数据分区（role='activeA'）

在某个时候，切换回“A”等。

最后一句话：如果Hive在每个压缩作业中仍然创建了太多的文件，那么在插入之前，尝试在会话中调整一些参数，例如。

set hive.merge.mapfiles =true;
set hive.merge.mapredfiles =true;
set hive.merge.smallfiles.avgsize=1024000000;