写入 Hive 表时的多个 Parquet 文件（增量）答案

【问题标题】：Multiple Parquet files while writing to Hive Table(Incremental)写入 Hive 表时的多个 Parquet 文件（增量）
【发布时间】：2018-12-28 13:44:51
【问题描述】：

拥有一个已分区的 Hive 表

CREATE EXTERNAL TABLE IF NOT EXISTS CUSTOMER_PART ( 
  NAME string ,
  AGE int ,
  YEAR INT)
  PARTITIONED BY (CUSTOMER_ID decimal(15,0))
  STORED AS PARQUET LOCATION 'HDFS LOCATION'

第一个 LOAD 是使用

通过 PYSPARK 从 ORACLE 到 HIVE 完成的

INSERT OVERWRITE TABLE CUSTOMER_PART PARTITION (CUSTOMER_ID) SELECT NAME, AGE, YEAR, CUSTOMER_ID FROM CUSTOMER;

它工作正常并在运行期间动态创建分区。现在每天增量加载数据会为分区下的单个记录创建单独的文件。

INSERT INTO TABLE CUSTOMER_PART PARTITION (CUSTOMER_ID = 3) SELECT NAME, AGE, YEAR FROM CUSTOMER WHERE CUSTOMER_ID = 3; --Assume this gives me the latest record in the database

是否有可能将该值附加到分区下的现有 parquet 文件中，直到达到其块大小，而无需为每个插入创建较小的文件。

重写整个分区是一种选择，但我不想这样做

INSERT OVERWRITE TABLE CUSTOMER_PART PARTITION (CUSTOMER_ID = 3) SELECT NAME, AGE, YEAR FROM CUSTOMER WHERE CUSTOMER_ID = 3;

为 Hive 设置了以下属性

set hive.execution.engine=tez; -- TEZ execution engine
set hive.merge.tezfiles=true; -- Notifying that merge step is required
set hive.merge.smallfiles.avgsize=128000000; --128MB
set hive.merge.size.per.task=128000000; -- 128MB

这对日常插入仍然没有帮助。任何可以遵循的替代方法都会非常有帮助。

【问题讨论】：

“是否有可能将值附加到现有的 parquet 文件中” >> 否。 HDFS 存储不可变文件（带有附加/截断 CSV 文件的边缘情况）。诸如 Parquet 之类的列格式以一种复杂的方式存储它们的数据，并带有一个终止文件的“页脚”（带有用于连接现有文件片段和重建页脚的边缘情况）。从 Uber 工程博客中搜索“Hoodie”/“Hudi”（他们在某些时候更改了项目名称），以了解问题以及进行增量更新的复杂程度......跨度>

标签： hive pyspark pyspark-sql parquet

【解决方案1】：

据我所知，我们无法为每日分区数据存储单个文件，因为数据将由每天分区的不同部分文件存储。

由于您提到您正在从 Oracle DB 导入数据，因此您可以每次从 oracle DB 导入整个数据并覆盖到 HDFS。通过这种方式，您可以维护单个零件文件。

对于少量数据也不建议使用 HDFS。

【讨论】：

【解决方案2】：

对于这种情况，我可以想到以下方法：

方法1：

重新创建 Hive 表，即将增量数据加载到 CUSTOMER_PART 表之后。

创建一个 temp_CUSTOMER_PART 表，其中包含 CUSTOMER_PART 表数据的整个快照。
运行覆盖决赛桌CUSTOMER_PART从temp_CUSTOMER_PART表格中选择
在这种情况下，您将获得没有小文件的决赛桌。
注意您需要确保在创建临时表后没有新数据插入到 CUSTOMER_PART 表中。

方法2：

通过使用input_file_name() 函数：

检查每个分区中有多少个不同的文件名，然后只选择每个分区中包含超过10..etc 个文件的分区。
用这些分区创建一个temporary table，并且只创建一个选定的分区overwrite the final table。
注意您需要确保在创建临时表后没有新数据插入到 CUSTOMER_PART 表中，因为我们要覆盖决赛桌。

方法3：

Hive（不是 spark）提供覆盖并选择同一个表。即

insert overwrite table default.t1 partition(partiton_column) 
select * from default.t1; //overwrite and select from same t1 table

如果您遵循这种方式，那么一旦您的 spark 作业完成，就需要有 hive job triggered。
Hive 将在 running overwrite/select 同一个表时获取锁，因此如果有任何正在写入表的作业将等待。

另外： Orc format 将提供concatenate，它将合并小的 ORC 文件以创建一个新的更大的文件。 p>

 alter table <db_name>.<orc_table_name> [partition_column="val"] concatenate;

【讨论】：