如何在 MapReduce 作业中以 parquet 文件格式写入输出？答案

【问题标题】：How to write output in parquet fileformat in a MapReduce job?如何在 MapReduce 作业中以 parquet 文件格式写入输出？
【发布时间】：2018-02-23 10:23:55
【问题描述】：

我希望使用parquet-mr 库以 parquet 文件格式编写 MapReduce 输出，如下所示：

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

        FileInputFormat.addInputPath(job, new Path(args[1]));
        ParquetOutputFormat.setOutputPath(job, new Path(args[2]));
        ParquetOutputFormat.setCompression(job, CompressionCodecName.GZIP);


        SkipBadRecords.setMapperMaxSkipRecords(conf, Long.MAX_VALUE);
        SkipBadRecords.setAttemptsToStartSkipping(conf, 0);

        job.submit();

但是，我不断收到以下错误：

2018-02-23 09:32:58,325 WARN [main] org.apache.hadoop.mapred.YarnChild: Exception running child : java.lang.NullPointerException: writeSupportClass should not be null
    at org.apache.parquet.Preconditions.checkNotNull(Preconditions.java:38)
    at org.apache.parquet.hadoop.ParquetOutputFormat.getWriteSupport(ParquetOutputFormat.java:350)
    at org.apache.parquet.hadoop.ParquetOutputFormat.getRecordWriter(ParquetOutputFormat.java:293)
    at org.apache.parquet.hadoop.ParquetOutputFormat.getRecordWriter(ParquetOutputFormat.java:283)
    at org.apache.hadoop.mapred.ReduceTask$NewTrackingRecordWriter.<init>(ReduceTask.java:548)
    at org.apache.hadoop.mapred.ReduceTask.runNewReducer(ReduceTask.java:622)
    at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:390)
    at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:164)

我知道 writeSupportClass 需要传递/设置为类似

ParquetOutputFormat.setWriteSupportClass(job, ProtoWriteSupport.class);

但是我能问一下如何指定架构、实现 ProtoWriteSupport 或任何其他 WriteSupport 类吗？我需要实现哪些方法？有没有正确的示例？

如果有帮助，我的 MR 作业的输出应该类似于并以 parquet 格式存储：

Text      INTWRITABLE
 a            100

【问题讨论】：

ParquetOutputFormat.setWriteSupportClass() 已经存在。我添加了一个解决方案。
感谢您的建议！事实上，我正在寻找如何实现 ProtoWriteSupport 并根据我在 reducer 中的输出字段指定模式。为了更清楚起见，我已经编辑了我的问题。
我相信ProtoWriteSupport 已经足够通用并且应该可以使用了。用javadoc 链接更新了我的答案，检查嵌套类部分。
谢谢！这也有帮助 - cloudera.com/documentation/enterprise/5-4-x/topics/…

标签： hadoop mapreduce parquet file-format

【解决方案1】：

试试ParquetOutputFormat.setWriteSupportClass(job, ProtoWriteSupport.class);

ProtoWriteSupport<T extends MessageOrBuilder>

实现用于写入协议缓冲区的 WriteSupport。

检查Javadoc 以获取可用的嵌套默认类列表。

CDH Tutorial 介绍在 MapReduce、Hive、HBase 和 Pig 中使用 parquet 文件格式。

【讨论】：