MapReduce读写orc文件

博客地址:http://www.fanlegefan.com
文章地址:http://www.fanlegefan.com/archives/mapreduceorc/

MapReduce 读取ORC格式文件

创建orc格式hive表

createtable test_orc(name string,ageint) storedas orc

查看hive表结构

showcreatetable test_orcCREATETABLE`test_orc`(`name` string,`age`int)ROW FORMAT SERDE'org.apache.hadoop.hive.ql.io.orc.OrcSerde' 
STOREDAS INPUTFORMAT'org.apache.hadoop.hive.ql.io.orc.OrcInputFormat' 
OUTPUTFORMAT'org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat'
LOCATION'hdfs://localhost:9000/user/work/warehouse/test_orc'
TBLPROPERTIES ('transient_lastDdlTime'='1502868725')

插入测试数据

insertintotable test_orcselect name ,agefrom test limit10;

jar依赖

<dependency><groupId>org.apache.orc</groupId><artifactId>orc-core</artifactId><version>1.2.3</version></dependency><dependency><groupId>org.apache.orc</groupId><artifactId>orc-mapreduce</artifactId><version>1.1.0</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-mapreduce-client-core</artifactId><version>2.6.0</version></dependency>

MR读取ORC格式文件代码如下

packagecom.fan.hadoop.orc;

importcom.fan.hadoop.parquet.thrift.ParquetThriftWriterMR;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapreduce.OrcInputFormat;
import java.io.IOException;


public class OrcReaderMR {

    public static class OrcMap extends Mapper<NullWritable,OrcStruct,Text,IntWritable> {

        // Assume the ORC file has type: struct<s:string,i:int>
        public void map(NullWritable key, OrcStruct value,
                        Context output) throws IOException, InterruptedException {
            // take the first field as the keyand the second field as the value
            output.write((Text) value.getFieldValue(0),
                    (IntWritable) value.getFieldValue(1));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);
        job.setJarByClass(ParquetThriftWriterMR.class);
        job.setJobName("parquetthrfit");

        Stringin ="hdfs://localhost:9000/user/work/warehouse/test_orc";
        Stringout ="hdfs://localhost:9000/test/orc";

        job.setMapperClass(OrcMap.class);
        OrcInputFormat.addInputPath(job, new Path(in));
        job.setInputFormatClass(OrcInputFormat.class);
        job.setNumReduceTasks(0);

        job.setOutputFormatClass(TextOutputFormat.class);

        FileOutputFormat.setOutputPath(job, new Path(out));


        job.waitForCompletion(true);
    }

}

查看生成文件

hadoop dfs-cat /test/orc/part-m-00000

kafka14
tensflow98
hadoop34
hbase68
flume57
kafka99
kafka28
flume24
tensflow35
flume44

MR写ORC格式文件

packagecom.fan.hadoop.orc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.orc.OrcConf;
import org.apache.orc.TypeDescription;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapreduce.OrcOutputFormat;
import java.io.IOException;

public class OrcWriterMR {

    public static class OrcWriterMapper
            extends Mapper<LongWritable,Text,NullWritable,OrcStruct> {


        private TypeDescription schema =
                TypeDescription.fromString("struct<name:string,age:int>");

        private OrcStruct pair = (OrcStruct) OrcStruct.createValue(schema);


        private final NullWritable nada = NullWritable.get();
        private Text name = new Text();
        private IntWritable age = new IntWritable();

        public void map(LongWritable key, Text value,
                           Context output
        ) throws IOException, InterruptedException {

            if(!"".equals(value.toString())){
                String[] arr = value.toString().split("\t");
                name.set(arr[0]);
                age.set(Integer.valueOf(arr[1]));
                pair.setFieldValue(0, name);
                pair.setFieldValue(1,age);
                output.write(nada, pair);
            }

        }
    }



    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        OrcConf.MAPRED_OUTPUT_SCHEMA.setString(conf,"struct<name:string,age:int>");

        Job job = Job.getInstance(conf);
        job.setJarByClass(OrcWriterMR.class);
        job.setJobName("OrcWriterMR");

        Stringin ="hdfs://localhost:9000/user/work/warehouse/test/ddd.txt";
        Stringout ="hdfs://localhost:9000/test/orc2";


        job.setMapperClass(OrcWriterMapper.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setNumReduceTasks(0);

        job.setOutputFormatClass(OrcOutputFormat.class);
        FileInputFormat.addInputPath(job, new Path(in));

        OrcOutputFormat.setOutputPath(job, new Path(out));


        job.waitForCompletion(true);
    }
}

查看生成文件

####生成orc文件hadoopdfs-ls/test/orc2-rw-r--r--3worksupergroup02017-08-1617:45/test/orc2/_SUCCESS-rw-r--r--3worksupergroup63148742017-08-1617:45/test/orc2/part-m-00000.orc

将数据放到hive表路径下

hadoop fs-cp /test/orc2/part-m-00000.orc /user/work/warehouse/test_orc/

在hive表中查看数据

hive>select *from test_orc limit13;
OK
kafka14
tensflow98
hadoop34
hbase68
flume57
kafka99
kafka28
flume24
tensflow35
flume44
flume44
tensflow35
flume24
Time taken:0.045 seconds, Fetched:13 row(s)