代码之家  ›  专栏  ›  技术社区  ›  Juhong Jung

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

reducers hive hadoop
  •  2
  • Juhong Jung  · 技术社区  · 7 年前

    我在用蜂箱。

    使用INSERT query编写动态分区并启用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 ,但对我没有帮助。

    我的环境是:

    • 使用tez作为执行引擎
    • 大约有30个动态分区要用INSERT query编写 

    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
;
    2 回复  |  直到 5 年前
        1
  •  1
  •   leftjoin    7 年前

    distribute by partition key 有助于解决OOM问题,但此配置可能会导致每个reducer写入整个分区,具体取决于 hive.exec.reducers.bytes.per.reducer 按分区密钥分发 可能会导致额外的减少阶段,同样的 hive.optimize.sort.dynamic.partition .

    因此,要避免OOM并实现最大性能:

    1. 添加 在insert查询结束时,这将导致相同的分区键由相同的reducer处理。或者,除了此设置之外,您还可以使用 hive.optimize.sort.dynamic.partition=true
    2. 配置单元.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

        2
  •  1
  •   Juhong Jung    7 年前

    最后我发现了问题所在。

    首先,执行引擎是tez。 mapreduce.reduce.memory.mb hive.tez.container.size 选项。写动态分区时,reducer会打开多个记录编写器。Reducer需要足够的内存来同时写入多个分区。

    如果你使用 hive.optimize.sort.dynamic.partition 选项,则运行全局分区排序,但排序意味着存在缩减器。在这种情况下,如果没有其他reducer任务,则每个分区由一个reducer处理。这就是为什么分区中只有一个文件。通过makemore-reduce任务进行分发,这样可以在每个分区中生成更多的文件,但是内存问题是相同的。

    因此,容器内存大小非常重要!别忘了用 hive.tez.container.size配置单元 选择改变tez容器内存大小!

    推荐文章
    Victor Wong  ·  如何迭代列A以有条件地对列B求和?
    1 年前
    StripyTiger  ·  从配置单元中的列表检索元素
    2 年前
    mfcss  ·  SQL:使用年/月/日分区查询某个时间段的数据
    2 年前
    MOMO  ·  配置单元重命名具有分区的表中的列
    3 年前
    Minnu  ·  json结构类型加载到配置单元
    3 年前
    PPSATO  ·  替代在AWS Athena中使用last_value和first_value
    3 年前
    te time  ·  按组获取完整的行集
    3 年前
    MattRgx  ·  配置单元中的Regexp将一列拆分为3列
    4 年前
    mishalhaneef  ·  如何更新蜂箱颤振数据
    4 年前
    Bhanuchander Udhayakumar  ·  为什么spark sql的cpu利用率高于hive?
    8 年前
    关于   移动版
    代码之家 - 一站式码农服务社区
    沪ICP备11025650号