指定从 Hive 插入生成的文件的最小数量
Specify minimum number of generated files from Hive insert
我在 AWS EMR 上使用 Hive 将查询结果插入按日期分区的 Hive table。虽然每天的总输出大小相似，但生成的文件数量各不相同，通常在 6 到 8 个之间，但有时它只创建一个大文件。我重新运行了几次查询，以防文件数量恰好受到集群中节点可用性的影响，但它似乎是一致的。
所以我的问题是
(a) 什么决定了生成多少文件以及
(b) 有没有办法指定最小文件数或（甚至更好）每个文件的最大大小？
在 INSERT ... SELECT 期间生成的文件数量取决于最终减速器上的进程数 运行（如果您在 Tez 上 运行 则为最终减速器顶点）加上配置的每个减速器的字节数.
如果 table 被分区并且没有指定 DISTRIBUTE BY ，那么在最坏的情况下，每个 reducer 在每个分区中创建文件。这样会给reducer造成很大的压力，可能会导致OOM异常
为确保每个 reducer 只写入一个分区文件，请在查询末尾添加 DISTRIBUTE BY partition_column。
如果数据量太大，并且你想要更多的 reducer 来增加并行度并为每个分区创建更多的文件，请将随机数添加到分发方式中，例如使用：FLOOR(RAND()*100.0)%10 - 它会通过随机 10 个桶额外分发数据，因此每个分区中将有 10 个文件。
最后您的 INSERT 语句将如下所示：
INSERT OVERWRITE table PARTITION(part_col)
SELECT *
FROM src
DISTRIBUTE BY part_col, FLOOR(RAND()*100.0)%10; --10 files per partition
此配置设置还会影响生成的文件数：
set hive.exec.reducers.bytes.per.reducer=67108864;
如果数据太多，Hive 将启动更多的 reducer 以处理不超过每个 reducer 进程指定的 bytes per reducer。 reducer 越多，生成的文件就越多。降低此设置可能会导致 reducer 的数量增加 运行，并且它们将为每个 reducer 创建至少一个文件。如果分区列不在 distribute by 中，那么每个 reducer 都可以在每个分区中创建文件。
长话短说，使用
DISTRIBUTE BY part_col, FLOOR(RAND()*100.0)%10 -- 10 files per partition
如果你想每个分区有20个文件，使用FLOOR(RAND()*100.0)%20; - 如果您有足够的数据，这将保证每个分区至少有 20 个文件，但不能保证每个文件的最大大小。
每个减速器设置的字节数不保证它将是固定的最小文件数。文件的数量将取决于总数据 size/bytes.per.reducer。 此设置将保证每个文件的最大大小。
但是最好使用一些均匀分布的键或基数较低的组合而不是随机，因为在容器重新启动的情况下，rand() 可能会为相同的行和它可能会导致数据重复或丢失（某些减速器输出中已经存在的相同数据将再次分发到另一个减速器）。您可以在某些可用键而不是 rand() 上计算类似函数，以获得或多或少均匀分布的低基数键。
您可以结合使用这两种方法：每个 reducer 限制的字节数 + 分发依据来控制最小文件数和最大文件大小。
另请阅读有关使用 distribute by 在 reducer 之间平均分配数据的答案：
我在 AWS EMR 上使用 Hive 将查询结果插入按日期分区的 Hive table。虽然每天的总输出大小相似，但生成的文件数量各不相同，通常在 6 到 8 个之间，但有时它只创建一个大文件。我重新运行了几次查询，以防文件数量恰好受到集群中节点可用性的影响，但它似乎是一致的。
所以我的问题是 (a) 什么决定了生成多少文件以及 (b) 有没有办法指定最小文件数或（甚至更好）每个文件的最大大小？
在 INSERT ... SELECT 期间生成的文件数量取决于最终减速器上的进程数 运行（如果您在 Tez 上 运行 则为最终减速器顶点）加上配置的每个减速器的字节数.
如果 table 被分区并且没有指定 DISTRIBUTE BY ，那么在最坏的情况下，每个 reducer 在每个分区中创建文件。这样会给reducer造成很大的压力，可能会导致OOM异常
为确保每个 reducer 只写入一个分区文件，请在查询末尾添加 DISTRIBUTE BY partition_column。
如果数据量太大，并且你想要更多的 reducer 来增加并行度并为每个分区创建更多的文件，请将随机数添加到分发方式中，例如使用：FLOOR(RAND()*100.0)%10 - 它会通过随机 10 个桶额外分发数据，因此每个分区中将有 10 个文件。
最后您的 INSERT 语句将如下所示：
INSERT OVERWRITE table PARTITION(part_col)
SELECT *
FROM src
DISTRIBUTE BY part_col, FLOOR(RAND()*100.0)%10; --10 files per partition
此配置设置还会影响生成的文件数：
set hive.exec.reducers.bytes.per.reducer=67108864;
如果数据太多，Hive 将启动更多的 reducer 以处理不超过每个 reducer 进程指定的 bytes per reducer。 reducer 越多，生成的文件就越多。降低此设置可能会导致 reducer 的数量增加 运行，并且它们将为每个 reducer 创建至少一个文件。如果分区列不在 distribute by 中，那么每个 reducer 都可以在每个分区中创建文件。
长话短说，使用
DISTRIBUTE BY part_col, FLOOR(RAND()*100.0)%10 -- 10 files per partition
如果你想每个分区有20个文件，使用FLOOR(RAND()*100.0)%20; - 如果您有足够的数据，这将保证每个分区至少有 20 个文件，但不能保证每个文件的最大大小。
每个减速器设置的字节数不保证它将是固定的最小文件数。文件的数量将取决于总数据 size/bytes.per.reducer。 此设置将保证每个文件的最大大小。
但是最好使用一些均匀分布的键或基数较低的组合而不是随机，因为在容器重新启动的情况下，rand() 可能会为相同的行和它可能会导致数据重复或丢失（某些减速器输出中已经存在的相同数据将再次分发到另一个减速器）。您可以在某些可用键而不是 rand() 上计算类似函数，以获得或多或少均匀分布的低基数键。
您可以结合使用这两种方法：每个 reducer 限制的字节数 + 分发依据来控制最小文件数和最大文件大小。
另请阅读有关使用 distribute by 在 reducer 之间平均分配数据的答案：