Hadoop支持的文件格式之Parquet

阳光穿透心脏的1/2处 2021-09-23 13:40 940阅读 0赞

### 文章目录 ###

*  0x00 文章内容
 *  0x01 行存储与列存储
 *   *   *   *   *  1. Avro与Parquet
 *  0x02 编码实现Parquet格式的读写
 *   *   *   *   *  1. 编码实现读写Parquet文件
                 *  2. 查看读写Parquet文件结果
                 *  3. 编码实现读写Parquet文件（HDFS）
                 *  4. 查看读写Parquet文件（HDFS）结果
 *  0x03 彩蛋
 *  0xFF 总结

# 0x00 文章内容 #

1.  行存储与列存储
2.  编码实现Parquet格式的读写
3.  彩蛋

# 0x01 行存储与列存储 #

##### 1. Avro与Parquet #####

a. 请参考文章：[Hadoop支持的文件格式之Avro][Hadoop_Avro]的[0x01 行存储与列存储][0x01]

# 0x02 编码实现Parquet格式的读写 #

##### 1. 编码实现读写Parquet文件 #####

a. 引入Parquet相关jar包

<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>

b. 完整的写Parquet文件代码（写到HDFS）

package com.shaonaiyi.hadoop.filetype.parquet;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.parquet.column.ParquetProperties;
    import org.apache.parquet.example.data.Group;
    import org.apache.parquet.example.data.GroupFactory;
    import org.apache.parquet.example.data.simple.SimpleGroupFactory;
    import org.apache.parquet.hadoop.ParquetWriter;
    import org.apache.parquet.hadoop.example.GroupWriteSupport;
    import org.apache.parquet.hadoop.metadata.CompressionCodecName;
    import org.apache.parquet.schema.MessageType;
    import org.apache.parquet.schema.MessageTypeParser;
    
    import java.io.IOException;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/18 10:14 * @Description 编码实现写Parquet文件 */
    public class ParquetFileWriter { 
    
        public static void main(String[] args) throws IOException { 
            MessageType schema = MessageTypeParser.parseMessageType("message Person {\n" +
                    " required binary name;\n" +
                    " required int32 age;\n" +
                    " required int32 favorite_number;\n" +
                    " required binary favorite_color;\n" +
                    "}");
    
            Configuration configuration = new Configuration();
            Path path = new Path("hdfs://master:9999/user/hadoop-sny/mr/filetype/parquet/data.parquet");
            GroupWriteSupport writeSupport = new GroupWriteSupport();
            GroupWriteSupport.setSchema(schema, configuration);
            ParquetWriter<Group> writer = new ParquetWriter<Group>(path, writeSupport,
                    CompressionCodecName.SNAPPY,
                    ParquetWriter.DEFAULT_BLOCK_SIZE,
                    ParquetWriter.DEFAULT_PAGE_SIZE,
                    ParquetWriter.DEFAULT_PAGE_SIZE,
                    ParquetWriter.DEFAULT_IS_DICTIONARY_ENABLED,
                    ParquetWriter.DEFAULT_IS_VALIDATING_ENABLED,
                    ParquetProperties.WriterVersion.PARQUET_1_0, configuration);
    
            GroupFactory groupFactory = new SimpleGroupFactory(schema);
            Group group = groupFactory.newGroup()
                    .append("name", "shaonaiyi")
                    .append("age", 18)
                    .append("favorite_number", 7)
                    .append("favorite_color", "red");
    
            writer.write(group);
    
            writer.close();
        }
    
    }

c. 完整的读Parquet文件代码（从HDFS读）

package com.shaonaiyi.hadoop.filetype.parquet;
    
    import org.apache.hadoop.fs.Path;
    import org.apache.parquet.example.data.Group;
    import org.apache.parquet.hadoop.ParquetReader;
    import org.apache.parquet.hadoop.example.GroupReadSupport;
    
    import java.io.IOException;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/18 10:18 * @Description 编码实现读Parquet文件 */
    public class ParquetFileReader { 
    
        public static void main(String[] args) throws IOException { 
    
    
            Path path = new Path("hdfs://master:9999/user/hadoop-sny/mr/filetype/parquet/parquet-data.parquet");
            GroupReadSupport readSupport = new GroupReadSupport();
            ParquetReader<Group> reader = new ParquetReader<>(path, readSupport);
    
            Group result = reader.read();
            System.out.println("name：" + result.getString("name", 0).toString());
            System.out.println("age：" + result.getInteger("age", 0));
            System.out.println("favorite_number：" + result.getInteger("favorite_number", 0));
            System.out.println("favorite_color：" + result.getString("favorite_color", 0));
        }
    
    }

##### 2. 查看读写Parquet文件结果 #####

a. 写Parquet文件  
![在这里插入图片描述][20191218102947579.png]  
b. 读Parquet文件  
![在这里插入图片描述][20191218103155401.png]

##### 3. 编码实现读写Parquet文件（HDFS） #####

a. 引入Parquet与Avro关联的jar包

<dependency>
            <groupId>org.apache.parquet</groupId>
            <artifactId>parquet-avro</artifactId>
            <version>1.8.1</version>
        </dependency>

从上面的代码我们可以看出，以下面这种方式定义Schema很不友好：

MessageType schema = MessageTypeParser.parseMessageType("message Person {\n" +
                " required binary name;\n" +
                " required int32 age;\n" +
                " required int32 favorite_number;\n" +
                " required binary favorite_color;\n" +
                "}");

所以我们可以将Parquet与Avro关联，直接使用Avro的Schema即可。

b. 完整的写Parquet文件代码（HDFS）

package com.shaonaiyi.hadoop.filetype.parquet;
    
    import com.shaonaiyi.hadoop.filetype.avro.Person;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.mapreduce.*;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.task.JobContextImpl;
    import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
    import org.apache.parquet.avro.AvroParquetOutputFormat;
    
    import java.io.IOException;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/18 10:47 * @Description 编码实现写Parquet文件（HDFS） */
    public class MRAvroParquetFileWriter { 
    
        public static void main(String[] args) throws IOException, IllegalAccessException, InstantiationException, ClassNotFoundException, InterruptedException { 
            //1 构建一个job实例
            Configuration hadoopConf = new Configuration();
            Job job = Job.getInstance(hadoopConf);
    
            //2 设置job的相关属性
            job.setOutputKeyClass(NullWritable.class);
            job.setOutputValueClass(Person.class);
            job.setOutputFormatClass(AvroParquetOutputFormat.class);
            //AvroJob.setOutputKeySchema(job, Schema.create(Schema.Type.INT));
            AvroParquetOutputFormat.setSchema(job, Person.SCHEMA$);
    
    
            //3 设置输出路径
            FileOutputFormat.setOutputPath(job, new Path("hdfs://master:9999/user/hadoop-sny/mr/filetype/avro-parquet"));
    
            //4 构建JobContext
            JobID jobID = new JobID("jobId", 123);
            JobContext jobContext = new JobContextImpl(job.getConfiguration(), jobID);
    
            //5 构建taskContext
            TaskAttemptID attemptId = new TaskAttemptID("attemptId", 123, TaskType.REDUCE, 0, 0);
            TaskAttemptContext hadoopAttemptContext = new TaskAttemptContextImpl(job.getConfiguration(), attemptId);
    
            //6 构建OutputFormat实例
            OutputFormat format = job.getOutputFormatClass().newInstance();
    
            //7 设置OutputCommitter
            OutputCommitter committer = format.getOutputCommitter(hadoopAttemptContext);
            committer.setupJob(jobContext);
            committer.setupTask(hadoopAttemptContext);
    
            //8 获取writer写数据，写完关闭writer
            RecordWriter<Void, Person> writer = format.getRecordWriter(hadoopAttemptContext);
            Person person = new Person();
            person.setName("shaonaiyi");
            person.setAge(18);
            person.setFavoriteNumber(7);
            person.setFavoriteColor("red");
            writer.write(null, person);
            writer.close(hadoopAttemptContext);
    
            //9 committer提交job和task
            committer.commitTask(hadoopAttemptContext);
            committer.commitJob(jobContext);
        }
    
    }

c. 完整的读Parquet文件代码（HDFS）

package com.shaonaiyi.hadoop.filetype.parquet;
    
    import com.shaonaiyi.hadoop.filetype.avro.Person;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.mapreduce.*;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.task.JobContextImpl;
    import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
    import org.apache.parquet.avro.AvroParquetInputFormat;
    
    import java.io.IOException;
    import java.util.List;
    import java.util.function.Consumer;
    /** * @Author shaonaiyi@163.com * @Date 2019/12/18 10:52 * @Description 编码实现读Parquet文件（HDFS） */
    public class MRAvroParquetFileReader { 
    
        public static void main(String[] args) throws IOException, IllegalAccessException, InstantiationException { 
            //1 构建一个job实例
            Configuration hadoopConf = new Configuration();
    
            Job job = Job.getInstance(hadoopConf);
    
            //2 设置需要读取的文件全路径
            FileInputFormat.setInputPaths(job, "hdfs://master:9999/user/hadoop-sny/mr/filetype/avro-parquet");
    
            //3 获取读取文件的格式
            AvroParquetInputFormat inputFormat = AvroParquetInputFormat.class.newInstance();
    
            AvroParquetInputFormat.setAvroReadSchema(job, Person.SCHEMA$);
            //AvroJob.setInputKeySchema(job, Person.SCHEMA$);
    
            //4 获取需要读取文件的数据块的分区信息
            //4.1 获取文件被分成多少数据块了
            JobID jobID = new JobID("jobId", 123);
            JobContext jobContext = new JobContextImpl(job.getConfiguration(), jobID);
    
            List<InputSplit> inputSplits = inputFormat.getSplits(jobContext);
    
            //读取每一个数据块的数据
            inputSplits.forEach(new Consumer<InputSplit>() { 
                @Override
                public void accept(InputSplit inputSplit) { 
                    TaskAttemptID attemptId = new TaskAttemptID("jobTrackerId", 123, TaskType.MAP, 0, 0);
                    TaskAttemptContext hadoopAttemptContext = new TaskAttemptContextImpl(job.getConfiguration(), attemptId);
                    RecordReader<NullWritable, Person> reader = null;
                    try { 
                        reader = inputFormat.createRecordReader(inputSplit, hadoopAttemptContext);
                        reader.initialize(inputSplit, hadoopAttemptContext);
                        while (reader.nextKeyValue()) { 
                            System.out.println(reader.getCurrentKey());
                            Person person = reader.getCurrentValue();
                            System.out.println(person);
                        }
                        reader.close();
                    } catch (IOException e) { 
                        e.printStackTrace();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            });
    
        }
    
    }

##### 4. 查看读写Parquet文件（HDFS）结果 #####

a. 写Parquet文件（HDFS）  
![在这里插入图片描述][20191218105612400.png]  
b. 读Parquet文件（HDFS），Key没有设置值  
![在这里插入图片描述][20191218105444399.png]

# 0x03 彩蛋 #

1.  编写读写Parquet文件Demo

package com.shaonaiyi.hadoop.filetype.parquet;
    
    import com.shaonaiyi.hadoop.filetype.avro.Person;
    import org.apache.hadoop.fs.Path;
    import org.apache.parquet.avro.AvroParquetReader;
    import org.apache.parquet.avro.AvroParquetWriter;
    import org.apache.parquet.hadoop.ParquetReader;
    import org.apache.parquet.hadoop.ParquetWriter;
    import org.apache.parquet.hadoop.metadata.CompressionCodecName;
    
    import java.io.IOException;
    
    /** * @Author shaonaiyi@163.com * @Date 2019/12/18 11:11 * @Description 编写读写Parquet文件Demo */
    public class AvroParquetDemo { 
    
        public static void main(String[] args) throws IOException { 
            Person person = new Person();
            person.setName("shaonaiyi");
            person.setAge(18);
            person.setFavoriteNumber(7);
            person.setFavoriteColor("red");
    
            Path path = new Path("hdfs://master:9999/user/hadoop-sny/mr/filetype/avro-parquet2");
    
            ParquetWriter<Object> writer = AvroParquetWriter.builder(path)
                    .withSchema(Person.SCHEMA$)
                    .withCompressionCodec(CompressionCodecName.SNAPPY)
                    .build();
    
            writer.write(person);
    
            writer.close();
    
            ParquetReader<Object> avroParquetReader = AvroParquetReader.builder(path).build();
            Person record = (Person)avroParquetReader.read();
            System.out.println("name：" + record.getName());
            System.out.println("age：" + record.get("age").toString());
            System.out.println("favorite_number：" + record.get("favorite_number").toString());
            System.out.println("favorite_color：" + record.get("favorite_color"));
    
        }
    
    
    }

1.  控制台可以读出文件  
    ![在这里插入图片描述][20191218111700173.png]
2.  HDFS上也有数据了  
    ![在这里插入图片描述][20191218111634886.png]

# 0xFF 总结 #

1.  在MapReduce作业中如何使用：

job.setInputFormatClass(AvroParquetInputFormat.class);
        AvroParquetInputFormat.setAvroReadSchema(job, Person.SCHEMA$);
        
        job.setOutputFormatClass(ParquetOutputFormat.class);
        AvroParquetOutputFormat.setSchema(job, Person.SCHEMA$);

1.  文章：[网站用户行为分析项目之会话切割（二）][Link 1] 中 [9. 保存统计结果][9.] 时就是以Parquet的格式保存下来的。
2.  Hadoop支持的文件格式系列：  
    [Hadoop支持的文件格式之Text][Hadoop_Text]  
    [Hadoop支持的文件格式之Avro][Hadoop_Avro]  
    [Hadoop支持的文件格式之Parquet][Hadoop_Parquet]  
    [Hadoop支持的文件格式之SequenceFile][Hadoop_SequenceFile]

--------------------

**福利：**  
[邵奈一的技术博客导航][Link 7]  
[邵奈一][CSDN] 原创不易，如转载请标明出处。

--------------------

[Hadoop_Avro]: https://blog.csdn.net/shaock2018/article/details/103581591
[0x01]: https://blog.csdn.net/shaock2018/article/details/103581591#0x01__5
[20191218102947579.png]: /images/20210923/c1353d6b662f4abdba243eb7d3b96d6e.png
[20191218103155401.png]: /images/20210923/ad1def51454842f79ab940079dd1b3a9.png
[20191218105612400.png]: /images/20210923/7a2d1c2086884ccfb7601b8a2702c5f4.png
[20191218105444399.png]: /images/20210923/f9d1fd9751f549f79ad7d9ca44394817.png
[20191218111700173.png]: /images/20210923/e4044c8cbf114c71968ef54f4735aa1f.png
[20191218111634886.png]: /images/20210923/b0730c357ff84b02892ab1ce9336e5f2.png
[Link 1]: https://blog.csdn.net/shaock2018/article/details/100775214
[9.]: https://blog.csdn.net/shaock2018/article/details/100775214#9__328
[Hadoop_Text]: https://blog.csdn.net/shaock2018/article/details/103576005
[Hadoop_Parquet]: https://blog.csdn.net/shaock2018/article/details/103592425
[Hadoop_SequenceFile]: https://blog.csdn.net/shaock2018/article/details/103629726
[Link 2]: https://blog.csdn.net/shaock2018/article/details/91184254
[Link 3]: https://img-blog.csdnimg.cn/20190608000417972.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoYW9jazIwMTg=,size_16,color_FFFFFF,t_70
[Link 4]: https://img-blog.csdnimg.cn/20190514084602748.png
[Link 5]: https://weibo.com/5680168183/
[CSDN]: https://blog.csdn.net/shaock2018
[Link 6]: https://www.jianshu.com/u/6f033c6a3b90
[Link 7]: https://blog.csdn.net/shaock2018/article/details/89349249