Parquet 之mapreduce

在mapreduce中使用Parquet，根据不同的序列化方式，有不同的选择，下面以Avro为例：
使用 AvroParquetInputFormat 和 AvroParquetOutputFormat

@Override
    public int run(String[] strings)throws Exception {


        Path inputPath =new Path(strings[0]);
        Path outputPath =new Path(strings[1]);

        Job job = Job.getInstance(getConf(),"AvroParquetMapReduce");
        job.setJarByClass(getClass());

        job.setInputFormatClass(AvroParquetInputFormat.class);
        AvroParquetInputFormat.setInputPaths(job,inputPath);

        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setOutputFormatClass(AvroParquetOutputFormat.class);
        FileOutputFormat.setOutputPath(job,outputPath);
        AvroParquetOutputFormat.setSchema(job,StockAvg.SCHEMA$);return job.waitForCompletion(true) ?0 :1;
    }

    staticclassMapextendsMapper<Void,Stock,Text,DoubleWritable>{@Overrideprotected void map(Void key, Stock value, Context context)throws IOException, InterruptedException {
            context.write(new Text(value.getSymbol().toString()),new DoubleWritable(value.getOpen()));
        }
    }

    staticclassReduceextendsReducer<Text,DoubleWritable,Void,StockAvg> {@Overrideprotected void reduce(Text key, Iterable<DoubleWritable> values, Context context)throws IOException, InterruptedException {
            Mean mean =new Mean();for (DoubleWritableval :values){
                mean.increment(val.get());
            }

            StockAvg avg =new StockAvg();
            avg.setSymbol(key.toString());
            avg.setAvg(mean.getResult());
            context.write(null,avg);
        }
    }

这里的输入输出都是 Parquet文件。如果向输入是Text文件，只要不设置InputFormatClass即可。

如果改变input schema文件，Avro不能加载具体的class，会强制使用GenericData代替。

publicclassAvroProjectionParquetMapReduceextendsConfiguredimplementsTool {publicstaticvoid main(String[] args) throws Exception {
        args =new String[2];

        args[0] ="hdfs://hadoop:9000/user/madong/parquet-input";
        args[1] ="hdfs://hadoop:9000/user/madong/parquet-output";int code = ToolRunner.run(new AvroProjectionParquetMapReduce(),args);
        System.exit(code);
    }

    @Overridepublicint run(String[] strings) throws Exception {
        Path inputPath =new Path(strings[0]);
        Path outputPath =new Path(strings[1]);


        Job job = Job.getInstance(getConf(),"AvroProjectionParquetMapReduce");
        job.setJarByClass(AvroProjectionParquetMapReduce.class);

        job.setInputFormatClass(AvroParquetInputFormat.class);
        AvroParquetInputFormat.setInputPaths(job, inputPath);// predicate pushdown
        AvroParquetInputFormat.setUnboundRecordFilter(job, GoogleStockFilter.class);// projection pushdown
        Schema projection = Schema.createRecord(Stock.SCHEMA$.getName(),
                Stock.SCHEMA$.getDoc(), Stock.SCHEMA$.getNamespace(),false);
        List<Schema.Field> fields = Lists.newArrayList();for (Schema.Field field : Stock.SCHEMA$.getFields()) {if ("symbol".equals(field.name()) ||"open".equals(field.name())) {
                fields.add(new Schema.Field(field.name(), field.schema(), field.doc(),
                        field.defaultValue(), field.order()));
            }
        }
        projection.setFields(fields);
        AvroParquetInputFormat.setRequestedProjection(job, projection);


        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

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

        job.setOutputFormatClass(AvroParquetOutputFormat.class);
        FileOutputFormat.setOutputPath(job, outputPath);
        AvroParquetOutputFormat.setSchema(job, StockAvg.SCHEMA$);return job.waitForCompletion(true) ?0 :1;
    }publicstaticclassGoogleStockFilterimplementsUnboundRecordFilter {privatefinal UnboundRecordFilter filter;public GoogleStockFilter() {
            filter = ColumnRecordFilter.column("symbol", ColumnPredicates.equalTo("GOOG"));
        }

        @Overridepublic RecordFilter bind(Iterable<ColumnReader> readers) {return filter.bind(readers);
        }
    }staticclassMapextendsMapper<Void,Stock,Text,DoubleWritable> {

        @Overrideprotectedvoid map(Void key, Stock value, Context context) throws IOException, InterruptedException {if (value !=null) {
                context.write(new Text(value.getSymbol().toString()),new DoubleWritable(value.getOpen()));
            }
        }
    }staticclassReduceextendsReducer<Text,DoubleWritable,Void,StockAvg> {

        @Overrideprotectedvoid reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
            Mean mean =new Mean();for (DoubleWritable val :values){
                mean.increment(val.get());
            }

            StockAvgavg =new StockAvg();avg.setSymbol(key.toString());avg.setAvg(mean.getResult());
            context.write(null,avg);
        }
    }
}