指定从 Hive 插入生成的文件的最小数量

Question

我在 AWS EMR 上使用 Hive 将查询结果插入到按日期分区的 Hive 表中。虽然每天的总输出大小相似，但生成的文件数量各不相同，通常在 6 到 8 个之间，但有时它只会创建一个大文件。我重新运行了几次查询，以防万一文件数量碰巧受到集群中节点可用性的影响，但它似乎是一致的。

所以我的问题是（a）是什么决定了生成多少文件，（b）有没有办法指定最小文件数或（甚至更好）每个文件的最大大小？

Answer 1

期间生成的文件数量INSERT ... SELECT取决于在最终减速器上运行的进程数（如果您在 Tez 上运行，则为最终减速器顶点）加上配置的每个减速器的字节数。

如果表被分区并且没有DISTRIBUTE BY指定，那么在最坏的情况下每个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 来处理不超过bytes per 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 个文件，但不能保证每个文件的最大大小。

每个减速器设置的字节数不保证它将是固定的最小文件数。文件数量取决于总数据大小/bytes.per.reducer。此设置将保证每个文件的最大大小。

但是最好使用一些均匀分布的键或低基数的组合而不是随机的，因为在容器重启的情况下，rand() 可能会为相同的行产生不同的值，并且可能导致数据重复或丢失（相同的数据已经存在于一些减速器输出将再次分配给另一个减速器）。您可以在一些可用的键上计算类似的函数，而不是rand()获得或多或少均匀分布的具有低基数的键。

您可以结合使用这两种方法：每个减速器限制的字节数 + 分发方式来控制最小文件数和最大文件大小。

另请阅读有关使用distribute by在减速器之间均匀分布数据的答案：https : //stackoverflow.com/a/38475807/2700344