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

1、Hive支持

创建表时指定parquet格式即可:

create table tmp.orc_test(id bigint, name string, age int) stored as parquet TBLPROPERTIES('orc.compresssion'='SNAPPY')

压缩格式有"SNAPPY"和 "GZIP"两种,需要哪种格式指定即可。

2、SPARK支持

Spark读:
df  = spark.read.parquet("/tmp/test/orc_data")  # 读出来的数据是一个dataframe

Spark写:
df.write.format("parquet").save("/tmp/test/orc_data2")

3、Hadoop Streaming支持

hadoop jar /usr/local/hadoop-2.7.0/share/hadoop/tools/lib/hadoop-streaming-2.7.0.jar \
-libjars parquet_test.jar,hadoop2-iow-lib.jar,/usr/local/spark-2.1.0-bin-hadoop2.7/jars/parquet-column-1.8.1.jar,/usr/local/spark-2.1.0-bin-hadoop2.7/jars/parquet-common-1.8.1.jar,/usr/local/spark-2.1.0-bin-hadoop2.7/jars/parquet-encoding-1.8.1.jar,/usr/local/spark-2.1.0-bin-hadoop2.7/jars/parquet-hadoop-1.8.1.jar,/usr/local/spark-2.1.0-bin-hadoop2.7/jars/parquet-format-2.3.0-incubating.jar \
-D mapred.job.name="test_streaming" \
-D iow.streaming.output.schema="message example {required binary age;required binary name;required binary desc;}"  \
-D mapreduce.output.fileoutputformat.compress=true \
-D parquet.compression=gzip \
-D parquet.read.support.class=net.iponweb.hadoop.streaming.parquet.GroupReadSupport \
-D parquet.write.support.class=net.iponweb.hadoop.streaming.parquet.GroupWriteSupport \
-inputformat net.iponweb.hadoop.streaming.parquet.ParquetAsTextInputFormat \
-outputformat net.iponweb.hadoop.streaming.parquet.ParquetAsTextOutputFormat \
-input "/tmp/test/parquet_test"  \
-output "/tmp/test/streaming_parquet_test" \
 -mapper /bin/cat -reducer /bin/cat

外部包:https://github.com/whale2/iow-hadoop-streaming 原本想用1.8的parquet格式,后面发现1.8parquet的读写的数据格式是mapreduce包下面的api,hadoop streaming只能用mapred包下面的api。

class org.apache.parquet.hadoop.ParquetInputFormat not org.apache.hadoop.mapred.InputFormat

4、MapReduce支持

pom.xml

    <dependencies>
        <!-- https://mvnrepository.com/artifact/com.twitter/parquet-hadoop -->
        <dependency>
            <groupId>org.apache.parquet</groupId>
            <artifactId>parquet-common</artifactId>
            <version>1.8.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.parquet</groupId>
            <artifactId>parquet-encoding</artifactId>
            <version>1.8.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.parquet</groupId>
            <artifactId>parquet-column</artifactId>
            <version>1.8.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.parquet</groupId>
            <artifactId>parquet-hadoop</artifactId>
            <version>1.8.1</version>
        </dependency>


        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>2.7.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-common -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.7.0</version>
        </dependency>

    </dependencies>
package is.parquet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.parquet.example.data.Group;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.hadoop.ParquetInputFormat;
import org.apache.parquet.hadoop.ParquetOutputFormat;
import org.apache.parquet.hadoop.example.GroupReadSupport;
import org.apache.parquet.hadoop.example.GroupWriteSupport;

import java.io.IOException;
import java.util.StringTokenizer;


public class ParquetRWMR extends Configured implements Tool {

    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();;
        String writeSchema = "message example {\n" +
                "required binary id;\n" +
                "required binary name;\n" +
                "required binary des;\n" +
                "}";
        conf.set("parquet.example.schema",writeSchema);

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

        String in = "/tmp/test/parquet_test";
        String out = "/tmp/test/parquet_test_mr";

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputValueClass(Group.class);

        job.setMapperClass(WordCountMap.class);
        job.setReducerClass(WordCountReduce.class);

        job.setInputFormatClass(ParquetInputFormat.class);
        job.setOutputFormatClass(ParquetOutputFormat.class);

        ParquetInputFormat.setInputPaths(job,new Path(in));
        ParquetInputFormat.setReadSupportClass(job, GroupReadSupport.class);

        ParquetOutputFormat.setOutputPath(job, new Path(out));
        ParquetOutputFormat.setWriteSupportClass(job, GroupWriteSupport.class);

        boolean rt =job.waitForCompletion(true);
        return rt?0:1;
    }

    public static class WordCountMap extends
            Mapper<Void, Group, Text, Text> {

        private Text word = new Text();

        public void map(Void key, Group value, Context context)
                throws IOException, InterruptedException {
            Long first = value.getLong("0",0); //value.getLong方法第一个参数是字段名,如果该参数是key-value类型的,第二个参数传0即可。因为根据key返回的值是一个list,0即是取第一个
            String sec = value.getString("1",0);
            String third = value.getString("2",0);
            word.set(first.toString());
            context.write(word, new Text(sec + "\t" + third));
        }
    }

    public static class WordCountReduce extends
            Reducer<Text, Text, Void, Group> {
        private SimpleGroupFactory factory;

        public void reduce(Text key, Iterable<Text> values,
                           Context context) throws IOException, InterruptedException {
            int sum = 0;
            StringBuilder str = new StringBuilder();
            for (Text val : values) {
                String tmp_file[] = val.toString().split("\t");
                Group group = factory.newGroup()
                        .append("id",  key.toString())
                        .append("name", tmp_file[0])
                        .append("des",tmp_file[1]);
                context.write(null,group);
                break;
            }


        }

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            super.setup(context);
            factory = new SimpleGroupFactory(GroupWriteSupport.getSchema(context.getConfiguration()));

        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        int retnum = ToolRunner.run(conf,new ParquetRWMR(),args);
    }
}

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏更流畅、简洁的软件开发方式

我的数据访问函数库的源代码(三)——返回结构数组

/* 2008 4 25 更新 */ 我的数据访问函数库的源码。整个类有1400行,原先就是分开来写的,现在更新后还是分开来发一下吧。 第三部分:返回结构 ...

2076
来自专栏chenssy

【死磕Sharding-jdbc】---group by结果合并(2)

在sharding-jdbc源码之group by结果合并(1)中主要分析了sharding-jdbc如何在GroupByStreamResultSetMerg...

1032
来自专栏数据之美

Pig、Hive、MapReduce 解决分组 Top K 问题

问题: 有如下数据文件 city.txt (id, city, value) cat city.txt  1 wh 500 2 bj 600 3 wh 1...

2697
来自专栏函数式编程语言及工具

浅谈Slick(1)- 基本功能描述

   Slick (Scala language-integrated connection kit)是scala的一个FRM(Functional Relat...

1837
来自专栏草根专栏

使用xUnit为.net core程序进行单元测试(2)

下面有一点点内容是重叠的.... String Assert 测试string是否相等: [Fact] public void ...

5457
来自专栏LhWorld哥陪你聊算法

Hadoop源码篇---解读Mapprer源码outPut输出

上次讲完MapReduce的输入后,这次开始讲MapReduce的输出。注意MapReduce的原语很重要:

1363
来自专栏三丰SanFeng

扩展mysql - 手把手教你写udf

使用过MySQL的人都知道,MySQL有很多内置函数提供给使用者,包括字符串函数、数值函数、日期和时间函数等,给开发人员和使用者带来了很多方便。

6615
来自专栏个人分享

MapReduce编程实现学习

MapReduce主要包括两个阶段:一个是Map,一个是Reduce. 每一步都有key-value对作为输入和输出。

2205
来自专栏逆向技术

16位汇编第七讲汇编指令详解第第三讲

                             16位汇编第六讲汇编指令详解第第三讲 1.十进制调整指令 1. 十进制数调整指令对二进制运算的结果进行...

1905
来自专栏风口上的猪的文章

.NET面试题系列[14] - LINQ to SQL与IQueryable

"理解IQueryable的最简单方式就是,把它看作一个查询,在执行的时候,将会生成结果序列。" - Jon Skeet

1461

扫码关注云+社区

领取腾讯云代金券