避免使用 hive.optimize.sort.dynamic.partition 选项的单个文件答案

【问题标题】：Avoid single file with hive.optimize.sort.dynamic.partition option避免使用 hive.optimize.sort.dynamic.partition 选项的单个文件
【发布时间】：2019-03-25 15:26:28
【问题描述】：

我正在使用蜂巢。

当我使用 INSERT 查询编写动态分区并打开 hive.optimize.sort.dynamic.partition 选项（SET hive.optimize.sort.dynamic.partition=true）时，每个分区中始终只有一个文件。

但是如果我打开那个选项（SET hive.optimize.sort.dynamic.partition=false），我就会出现这样的内存不足异常。

TaskAttempt 3 failed, info=[Error: Error while running task ( failure ) : attempt_1534502930145_6994_1_01_000008_3:java.lang.RuntimeException: java.lang.OutOfMemoryError: Java heap space
        at org.apache.hadoop.hive.ql.exec.tez.TezProcessor.initializeAndRunProcessor(TezProcessor.java:194)
        at org.apache.hadoop.hive.ql.exec.tez.TezProcessor.run(TezProcessor.java:168)
        at org.apache.tez.runtime.LogicalIOProcessorRuntimeTask.run(LogicalIOProcessorRuntimeTask.java:370)
        at org.apache.tez.runtime.task.TaskRunner2Callable$1.run(TaskRunner2Callable.java:73)
        at org.apache.tez.runtime.task.TaskRunner2Callable$1.run(TaskRunner2Callable.java:61)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.Subject.doAs(Subject.java:422)
        at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
        at org.apache.tez.runtime.task.TaskRunner2Callable.callInternal(TaskRunner2Callable.java:61)
        at org.apache.tez.runtime.task.TaskRunner2Callable.callInternal(TaskRunner2Callable.java:37)
        at org.apache.tez.common.CallableWithNdc.call(CallableWithNdc.java:36)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.OutOfMemoryError: Java heap space
        at org.apache.parquet.column.values.dictionary.IntList.initSlab(IntList.java:90)
        at org.apache.parquet.column.values.dictionary.IntList.<init>(IntList.java:86)
        at org.apache.parquet.column.values.dictionary.DictionaryValuesWriter.<init>(DictionaryValuesWriter.java:93)
        at org.apache.parquet.column.values.dictionary.DictionaryValuesWriter$PlainBinaryDictionaryValuesWriter.<init>(DictionaryValuesWriter.java:229)
        at org.apache.parquet.column.ParquetProperties.dictionaryWriter(ParquetProperties.java:131)
        at org.apache.parquet.column.ParquetProperties.dictWriterWithFallBack(ParquetProperties.java:178)
        at org.apache.parquet.column.ParquetProperties.getValuesWriter(ParquetProperties.java:203)
        at org.apache.parquet.column.impl.ColumnWriterV1.<init>(ColumnWriterV1.java:83)
        at org.apache.parquet.column.impl.ColumnWriteStoreV1.newMemColumn(ColumnWriteStoreV1.java:68)
        at org.apache.parquet.column.impl.ColumnWriteStoreV1.getColumnWriter(ColumnWriteStoreV1.java:56)
        at org.apache.parquet.io.MessageColumnIO$MessageColumnIORecordConsumer.<init>(MessageColumnIO.java:184)
        at org.apache.parquet.io.MessageColumnIO.getRecordWriter(MessageColumnIO.java:376)
        at org.apache.parquet.hadoop.InternalParquetRecordWriter.initStore(InternalParquetRecordWriter.java:109)
        at org.apache.parquet.hadoop.InternalParquetRecordWriter.<init>(InternalParquetRecordWriter.java:99)
        at org.apache.parquet.hadoop.ParquetRecordWriter.<init>(ParquetRecordWriter.java:100)
        at org.apache.parquet.hadoop.ParquetOutputFormat.getRecordWriter(ParquetOutputFormat.java:327)
        at org.apache.parquet.hadoop.ParquetOutputFormat.getRecordWriter(ParquetOutputFormat.java:288)
        at org.apache.hadoop.hive.ql.io.parquet.write.ParquetRecordWriterWrapper.<init>(ParquetRecordWriterWrapper.java:67)
        at org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat.getParquerRecordWriterWrapper(MapredParquetOutputFormat.java:128)
        at org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat.getHiveRecordWriter(MapredParquetOutputFormat.java:117)
        at org.apache.hadoop.hive.ql.io.HiveFileFormatUtils.getRecordWriter(HiveFileFormatUtils.java:286)
        at org.apache.hadoop.hive.ql.io.HiveFileFormatUtils.getHiveRecordWriter(HiveFileFormatUtils.java:271)
        at org.apache.hadoop.hive.ql.exec.FileSinkOperator.createBucketForFileIdx(FileSinkOperator.java:619)
        at org.apache.hadoop.hive.ql.exec.FileSinkOperator.createBucketFiles(FileSinkOperator.java:563)
        at org.apache.hadoop.hive.ql.exec.FileSinkOperator.createNewPaths(FileSinkOperator.java:867)
        at org.apache.hadoop.hive.ql.exec.FileSinkOperator.getDynOutPaths(FileSinkOperator.java:975)
        at org.apache.hadoop.hive.ql.exec.FileSinkOperator.process(FileSinkOperator.java:715)
        at org.apache.hadoop.hive.ql.exec.Operator.forward(Operator.java:897)
        at org.apache.hadoop.hive.ql.exec.SelectOperator.process(SelectOperator.java:95)
        at org.apache.hadoop.hive.ql.exec.tez.ReduceRecordSource$GroupIterator.next(ReduceRecordSource.java:356)
        at org.apache.hadoop.hive.ql.exec.tez.ReduceRecordSource.pushRecord(ReduceRecordSource.java:287)
        at org.apache.hadoop.hive.ql.exec.tez.ReduceRecordProcessor.run(ReduceRecordProcessor.java:317)
]], Vertex did not succeed due to OWN_TASK_FAILURE, failedTasks:1 killedTasks:299, Vertex vertex_1534502930145_6994_1_01 [Reducer 2] killed/failed due to:OWN_TASK_FAILURE]Vertex killed, vertexName=Map 1, vertexId=vertex_1534502930145_6994_1_00, diagnostics=[Vertex received Kill while in RUNNING state., Vertex did not succeed due to OTHER_VERTEX_FAILURE, failedTasks:0 killedTasks:27, Vertex vertex_1534502930145_6994_1_00 [Map 1] killed/failed due to:OTHER_VERTEX_FAILURE]DAG did not succeed due to VERTEX_FAILURE. failedVertices:1 killedVertices:1

我猜这个异常是因为 reducer 同时写入许多分区。但我找不到如何控制它。我关注了this article，但对我没有帮助。

我的环境是这样的：

AWS EMR 5.12.1
使用 tez 作为执行引擎
hive 版本是 2.3.2，tez 版本是 0.8.2
HDFS 块大小为 128MB
大约有 30 个动态分区要使用 INSERT 查询写入

这是我的示例查询。

SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.optimize.sort.dynamic.partition=true;
SET hive.exec.reducers.bytes.per.reducer=1048576;
SET mapred.reduce.tasks=300;
FROM raw_data
INSERT OVERWRITE TABLE idw_data
  PARTITION(event_timestamp_date)
  SELECT
    *
  WHERE 
    event_timestamp_date BETWEEN '2018-09-09' AND '2018-10-09' 
DISTRIBUTE BY event_timestamp_date
;

【问题讨论】：

this: SET mapred.reduce.tasks=300; - 它可能会覆盖 bytes.per.reducer 并强制使用 300 个减速器。启动了多少个减速器？
并按 event_timestamp_date 分发 + 多一列（与分区相关性不大）肯定会为每个分区创建多个文件
@leftjoin 抱歉回复晚了。我试过distribute by event_timestamp_date + one more additional column (not much correlated with partition) 和变量hive.exec.reducers.bytes.per.reducer 作为1024 到104857600，但分区中总是只有一个文件。例如，分区中有超过 10GB 的文件。我无法理解这种情况，因为有超过 1000 个减少任务（顶点）任务，但结果是一个文件。
供您参考，我尝试使用SET hive.optimize.sort.dynamic.partition=true;
我尝试了执行引擎作为mr，结果是一样的。

标签： hadoop hive hiveql reducers hive-configuration

【解决方案1】：

distribute by partition key 有助于解决 OOM 问题，但此配置可能会导致每个 reducer 写入整个分区，具体取决于 hive.exec.reducers.bytes.per.reducer 配置，默认情况下可以设置非常高的值，例如 1Gb。 distribute by partition key 可能会导致额外的 reduce 阶段，hive.optimize.sort.dynamic.partition 也是如此。

所以，为了避免 OOM 并达到最佳性能：

在插入查询的末尾添加distribute by partition key，这将导致相同的分区键由相同的reducer处理。或者，除了此设置之外，您还可以使用 hive.optimize.sort.dynamic.partition=true
将hive.exec.reducers.bytes.per.reducer 设置为将如果一个分区中的数据过多，则触发更多的 reducer。只需检查 hive.exec.reducers.bytes.per.reducer 的当前值并相应地减少或增加它以获得适当的减速器并行度。此设置将决定单个 reducer 将处理多少数据以及每个分区将创建多少文件。

例子：

set hive.exec.reducers.bytes.per.reducer=33554432;

insert overwrite table partition (load_date)
select * from src_table
distribute by load_date;

另请参阅有关控制映射器和减速器数量的答案：https://stackoverflow.com/a/42842117/2700344

【讨论】：

我设置了hive.optimize.sort.dynamic.partition=true 和hive.exec.reducers.bytes.per.reducer=1024 但分区中仍有单个文件（超过10GB）。更高的值如 1048576、10485760，更小的值 sush 因为 512 没有做任何改变。
@JuhongJung 也尝试使用分发，每个 reducer 的字节数。没有 hive.optimize.sort.dynamic.partition。将每个reducer的字节设置为这么小的值有什么用，10485760 = 10M，还不到一个块？
我尝试了SET hive.optimize.sort.dynamic.partition=false 和SET hive.exec.reducers.bytes.per.reducer=1048576 并使用DISTRIBUTE BY event_timestamp_date，但它导致内存不足异常。我在我的问题上添加了示例查询。请检查一下：）非常感谢！

【解决方案2】：

终于找到问题所在了。

首先，执行引擎是 tez。 mapreduce.reduce.memory.mb 选项没有帮助。您应该使用hive.tez.container.size 选项。在写入动态分区时，reducer 会打开多个记录写入器。 Reducer 需要足够的内存来同时写入多个分区。

如果您使用hive.optimize.sort.dynamic.partition 选项，则会运行全局分区排序，但排序意味着存在减速器。在这种情况下，如果没有其他 reducer 任务，则每个分区由一个 reducer 处理。这就是为什么分区中只有一个文件的原因。 DISTRIBUTE BY 做更多的reduce任务，所以每个分区可以做更多的文件，但是同样存在内存问题。

因此，容器内存大小非常重要！不要忘记使用hive.tez.container.size 选项来更改 tez 容器内存大小！

【讨论】：