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 条评论
登录 后参与评论

相关文章

来自专栏程序员维他命

iOS 代码规范

花了一个月的时间结合几篇博客和书籍写了这套 iOS 代码规范(具体参考底部的参考文献部分)。这套代码规范除了有仅适用于 iOS 开发的部分,还有其他的比较通用性...

2002
来自专栏技术博客

Entity Framework DataAnnotations

DataAnnotation 特性由.NET 3.5中引进,给.NET中的类提供了一种添加验证的方式。但是在EF中它又可以对映射关系进行控制,相比较Fluent...

933
来自专栏逍遥剑客的游戏开发

MPQ文件系统优化(续)

1725
来自专栏Spark生态圈

[spark] Shuffle Write解析 (Sort Based Shuffle)

从 Spark 2.0 开始移除了Hash Based Shuffle,想要了解可参考Shuffle 过程,本文将讲解 Sort Based Shuffle。

1452
来自专栏Spark生态圈

[spark] Shuffle Read解析 (Sort Based Shuffle)

本文将讲解shuffle Reduce部分,shuffle的下游Stage的第一个rdd是ShuffleRDD,通过其compute方法来获取上游Stage S...

1211
来自专栏IT大咖说

Getting started with Kotlin on Android

摘要 Kotlin是一门与Swift类似的静态类型JVM语言,由JetBrain设计开发并开源,与Java相比,Kotlin的语法更简洁,更具表达性,而且提供了...

2835
来自专栏IT可乐

mybatis源码解读(二)——构建Configuration对象

1962
来自专栏Android知识点总结

Java总结IO篇之File类和Properties类

打开颜色选择器 :读流I-->字符串分割-->字符串存入Map-->使用Map对象还原用户配置 修改配置时 :写流O-->创建Map对象-->字符...

1802
来自专栏coolblog.xyz技术专栏

自己动手实现一个简单的JSON解析器

JSON(JavaScript Object Notation) 是一种轻量级的数据交换格式。相对于另一种数据交换格式 XML,JSON 有着诸多优点。比如易读...

67619
来自专栏JMCui

Netty 系列六(编解码器).

    网络传输的单位是字节,如何将应用程序的数据转换为字节,以及将字节转换为应用程序的数据,就要说到到我们该篇介绍的编码器和解码器。

1071

扫码关注云+社区

领取腾讯云代金券