专栏首页YG小书屋Hadoop Streaming 读ORC文件

Hadoop Streaming 读ORC文件

【背景】

hadoop Streaming的处理流程是先通过inputFormat读出输入文件内容,将其传递mapper,再将mapper返回的key,value传给reducer,最后将reducer返回的值通过outputformat写入输出文件。 目前有个需求是通过hadoop streaming读取roc文件。使用正常的org.apache.orc.mapred.OrcInputFormat读orc文件时每行返回的值是:

null    {"name":"123","age":"456"}
null    {"name":"456","age":"789"}

返回这种数据的原因是OrcInputFormat读取文件返回的值是<NullWritable, OrcStruct>, NullWritable toString的返回值是null, OrcStruct toString的返回值是一个json串。 需要开发一个转换器,只返回OrcInputFormat返回的json串的value即可。即返回:

123 456
456 789

【重写InputFormat,单文件读取】

package is.orc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.*;
import org.apache.orc.TypeDescription;
import org.apache.orc.mapred.OrcInputFormat;
import org.apache.orc.mapred.OrcMapredRecordReader;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.Reader;
import org.apache.orc.Reader.Options;
import java.io.IOException;

public class OrcInputAsTextInputFormat extends org.apache.hadoop.mapred.FileInputFormat<Text, Text> {
    //真正读文件的还是OrcInputFormat
    protected OrcInputFormat<OrcStruct> orcInputFormat = new OrcInputFormat();

    public RecordReader<Text, Text> getRecordReader(InputSplit split, JobConf job, Reporter reporter) throws IOException {
        OrcMapredRecordReader realReader = (OrcMapredRecordReader) orcInputFormat.getRecordReader(split, job, reporter);
        return new TextRecordReaderWrapper
                (realReader);
    }

    public static boolean[] parseInclude(TypeDescription schema, String columnsStr) {
        return OrcInputFormat.parseInclude(schema, columnsStr);
    }

    public static void setSearchArgument(Configuration conf, SearchArgument sarg, String[] columnNames) {
        OrcInputFormat.setSearchArgument(conf, sarg, columnNames);
    }

    public static Options buildOptions(Configuration conf, Reader reader, long start, long length) {
        return OrcInputFormat.buildOptions(conf, reader, start, length);
    }

    protected static class TextRecordReaderWrapper implements RecordReader<Text, Text> {

        private OrcMapredRecordReader realReader;
        private OrcStruct orcVal ;
        private StringBuilder buffer;
        private final int numOfFields;
        public TextRecordReaderWrapper(OrcMapredRecordReader realReader) throws IOException{
            this.realReader = realReader;
            this.orcVal = (OrcStruct)realReader.createValue();
            this.buffer = new StringBuilder();
            this.numOfFields = this.orcVal.getNumFields();
        }

        public boolean next(Text key, Text value) throws IOException {
            // 将第一个字段作为key,剩余的字段以\t为分隔符组成字符串作为value
            if (realReader.next(NullWritable.get(), orcVal)){
                buffer.setLength(0); //清空buffer
                key.set(orcVal.getFieldValue(0).toString());
                //以\t为分隔符,组装返回值
                for(int i = 1; i < numOfFields; ++i) {
                    buffer.append("\t");
                    WritableComparable curField = orcVal.getFieldValue(i);
                    if (curField != null && ! curField.equals(NullWritable.get())){
                        buffer.append(curField.toString());
                    }

                }
                value.set(buffer.substring(1)); //去掉开始添加的\t
                return Boolean.TRUE;
            }

            return Boolean.FALSE;
        }

        public Text createKey() {
            return new Text();
        }

        public Text createValue() {
            return new Text();
        }

        public long getPos() throws IOException {
            return realReader.getPos();
        }

        public void close() throws IOException {
            realReader.close();
        }

        public float getProgress() throws IOException {
            return realReader.getProgress();
        }
    }
}

【多文件读取】

MapReduce在读数据的时候可以通过合并小文件的方式减少map个数,比如说CombineSequenceFileInputFormat。如果不合并小文件,可能出现map数过大的情况,资源消耗过多,且执行效率很慢。对应到orc格式时没找到官方提供的包,只能自己写一个。具体代码如下:

package is.orc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.CombineFileInputFormat;
import org.apache.hadoop.mapred.lib.CombineFileRecordReader;
import org.apache.hadoop.mapred.lib.CombineFileRecordReaderWrapper;
import org.apache.hadoop.mapred.lib.CombineFileSplit;

import java.io.IOException;

public class CombineOrcInputAsTextInputFormat
        extends CombineFileInputFormat<org.apache.hadoop.io.Text, org.apache.hadoop.io.Text> {
    @SuppressWarnings({ "rawtypes", "unchecked" })
    public RecordReader<org.apache.hadoop.io.Text, org.apache.hadoop.io.Text> getRecordReader(InputSplit split, JobConf conf,
                                             Reporter reporter) throws IOException {
        return new CombineFileRecordReader(conf, (CombineFileSplit)split, reporter,
                ORCFileRecordReaderWrapper.class);
    }

    /**
     * A record reader that may be passed to <code>CombineFileRecordReader</code>
     * so that it can be used in a <code>CombineFileInputFormat</code>-equivalent
     * for <code>SequenceFileInputFormat</code>.
     *
     * @see CombineFileRecordReader
     * @see CombineFileInputFormat
     * @see SequenceFileInputFormat
     */
    private static class ORCFileRecordReaderWrapper
            extends CombineFileRecordReaderWrapper<org.apache.hadoop.io.Text, org.apache.hadoop.io.Text> {
        // this constructor signature is required by CombineFileRecordReader
        public ORCFileRecordReaderWrapper(CombineFileSplit split,
                                               Configuration conf, Reporter reporter, Integer idx) throws IOException {
            //只需配置此处的InputFormat为第一部分编写的OrcInputAsTextInputFormat即可。具体的合并操作,CombineFileInputFormat已帮我们实现
            super(new OrcInputAsTextInputFormat(), split, conf, reporter, idx);
        }
    }
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • parquet文件格式对常用系统的支持

    外部包:https://github.com/whale2/iow-hadoop-streaming 原本想用1.8的parquet格式,后面发现1.8par...

    YG
  • orc文件格式对常用系统的支持

    YG
  • MapReduce:N keys,N files

    MapReduce中,不管是map阶段还是reduce阶段,二者的输入和输出都是key,value类型的值。现在有个需求是根据map阶段返回值key的个数,生成...

    YG
  • mapreduce如何使用本地文件 转

    stys35
  • 干货--Hadoop自定义数据类型和自定义输入输出格式整合项目案例

    正文开始前 ,先介绍几个概念 序列化 所谓序列化,是指将结构化对象转化为字节流,以便在网络上传输或写到磁盘进行永久存储。 反序列化 是指将字节流转回到结构化...

    汤高
  • MapReduce Join

    王知无
  • MapReduce Join

    用户1621453
  • 大数据技术之_05_Hadoop学习_04_MapReduce_Hadoop企业优化(重中之重)+HDFS小文件优化方法+MapReduce扩展案例+倒排索引案例(多job串联)+TopN案例+找博客

      MapReduce优化方法主要从六个方面考虑:数据输入、Map阶段、Reduce阶段、IO传输、数据倾斜问题和常用的调优参数。

    黑泽君
  • hadoop_eclipse及HDT插件的使用

    甜橙很酸
  • MapReduce并行编程模型和框架

    Combiner:进行中间结果数据网络传输优化的工作。Combiner程序的执行是在Map节点完成计算之后、输出结果之前。

    chaplinthink

扫码关注云+社区

领取腾讯云代金券