Mapreduce读写Avro序列化文件

爱被打了一巴掌 2022-04-16 05:27 245阅读 0赞

**目录**

(1) MapReduce读取Avro格式文件

(2) MapReduce写入Avro格式文件

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

以WordCount为例来简单说明一下MapReduce如何读写Avro格式的序列化文件，小白初学，如有错误，欢迎批评指正！

首先pom.xml文件中需要引入相关依赖

<properties>
        <arvo.version>1.8.2</arvo.version>
    </properties>
    <dependencies>
        
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>${arvo.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-ipc</artifactId>
            <version>${arvo.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-mapred</artifactId>
            <version>${arvo.version}</version>
        </dependency>
    
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-mapred</artifactId>
            <version>${arvo.version}</version>
        </dependency>
        <dependency>
            <groupId>org.codehaus.jackson</groupId>
            <artifactId>jackson-core-asl</artifactId>
            <version>1.9.9</version>
        </dependency>
    </dependencies>

引入如下插件

<plugin>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro-maven-plugin</artifactId>
        <version>${arvo.version}</version>
        <executions>
            <execution>
                <id>schemas</id>
                <phase>generate-sources</phase>
                <goals>
                    <goal>schema</goal>
                </goals>
                <configuration>
                    <sourceDirectory>${project.basedir}/src/main/resources</sourceDirectory>
                    <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
                </configuration>
            </execution>
        </executions>
    </plugin>

### (1) MapReduce读取Avro格式文件 ###

准备好一个[Avro文件][Avro](可下载)，文件中的内容如下：

{"word":"China","count":1}
    {"word":"hello","count":3}
    {"word":"kitty","count":1}
    {"word":"to","count":1}
    {"word":"tom","count":1}
    {"word":"welcome","count":1}
    {"word":"world","count":1}

注：key为”word”的代表单词，key为”count”的代表单词的个数

其Schema为

{
        "namespace":"test.avro",
        "type":"record",
        "name":"read_write",
        "fields":[
            {
                "name":"word",
                "type":"string"
            },
            {
                "name":"count",
                "type":"int"
            }
        ]
    }

Mapper主要负责从文件中读取每个“word”对应的单词，然后以<word, 1>的形式传给Reducer

import org.apache.avro.generic.GenericRecord;
    import org.apache.avro.mapred.AvroKey;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    /**
     * @author YF_Li
     * @since 2018/11/17
     */
    public class AvroReaderMapper extends Mapper<AvroKey<GenericRecord>, NullWritable, Text, LongWritable> {
        private Text outputKey = new Text();
        private final LongWritable outputVal = new LongWritable(1);
    
        @Override
        protected void map(AvroKey<GenericRecord> key, NullWritable value, Context context) throws IOException, InterruptedException {
    
            // 从 Avro 文件中取出 'word' 对应的值
            String word = key.datum().get("word").toString();
            outputKey.set(word);
            context.write(outputKey,outputVal);
        }
    }

### (2) MapReduce写入Avro格式文件 ###

Reducer主要处理Mapper传过来的每个<word, 1>，对其value值进行求和。Mapper在将数据传给Reducer的时候会将key相同的进行合并，比如Mapper读取了两个“word”为“hello”的记录，则在Reducer中是的<hello, \[1,1\]>

import org.apache.avro.Schema;
    import org.apache.avro.generic.GenericData;
    import org.apache.avro.generic.GenericRecord;
    import org.apache.avro.mapred.AvroKey;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /**
     * @author YF_Li
     * @since 2018/11/17
     */
    public class AvroWriterReducer extends Reducer<Text, LongWritable, AvroKey<GenericRecord>, NullWritable> {
    
        // 定义 Avro 序列化文件 Schema
        private static final String SCHEMA = "{\n" +
                "  \"namespace\": \"test.avro\",\n" +
                "  \"type\": \"record\",\n" +
                "  \"name\": \"read_writer\",\n" +
                "  \"fields\": [{\n" +
                "    \"name\": \"word\",\n" +
                "    \"type\": \"string\"\n" +
                "  },{\n" +
                "    \"name\": \"count\",\n" +
                "    \"type\": \"int\"\n" +
                "  }]\n" +
                "}";
    
        LongWritable outputVal = new LongWritable();
    
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long count = 0L;
    
            // 循环values,对其值进行求和
            for (LongWritable value: values) {
                count += value.get();
            }
            outputVal.set(count);
    
            // 利用GenericRecord将数据写入Avro文件
            GenericRecord record = new GenericData.Record(new Schema.Parser().parse(SCHEMA));
            record.put("word", key);
            record.put("count", count);
    
            context.write(new AvroKey<GenericRecord>(record), NullWritable.get());
        }
    }

MapReduce Job 设置如下：

import org.apache.avro.Schema;
    import org.apache.avro.mapreduce.AvroJob;
    import org.apache.avro.mapreduce.AvroKeyInputFormat;
    import org.apache.avro.mapreduce.AvroKeyOutputFormat;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    /**
     * @author YF_Li
     * @since 2018/11/17
     */
    public class AvroReadWriteJob extends Configured implements Tool {
    
        // 定义 Avro 序列化文件 Schema
        private static final String SCHEMA = "{\n" +
                "  \"namespace\": \"test.avro\",\n" +
                "  \"type\": \"record\",\n" +
                "  \"name\": \"read_writer\",\n" +
                "  \"fields\": [{\n" +
                "    \"name\": \"word\",\n" +
                "    \"type\": \"string\"\n" +
                "  },{\n" +
                "    \"name\": \"count\",\n" +
                "    \"type\": \"int\"\n" +
                "  }]\n" +
                "}";
    
        @Override
        public int run(String[] strings) throws Exception {
    
            Configuration conf = getConf();
    
            // 获取输入输出路径
            String inputPath = conf.get("mapreduce.job.input_path");
            String outputPath = conf.get("mapreduce.job.output_path");
    
            Job job = Job.getInstance(conf);
            job.setJarByClass(AvroReadWriteJob.class);
            job.setMapperClass(AvroReaderMapper.class);
            job.setReducerClass(AvroWriterReducer.class);
    
            // 设置Mapper阶段的输出的 key、value 类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);
    
            // 设置Avro输出文件的Schema
            AvroJob.setOutputKeySchema(job, new Schema.Parser().parse(SCHEMA));
    
            // 设置输入文件和输出文件格式
            job.setInputFormatClass(AvroKeyInputFormat.class);
            job.setOutputFormatClass(AvroKeyOutputFormat.class);
    
            // 设置输入输出路径
            FileInputFormat.addInputPath(job, new Path(inputPath));
            FileOutputFormat.setOutputPath(job, new Path(outputPath));
    
            boolean res = job.waitForCompletion(true);
    
            return res ? 0 : 1;
        }
    
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            int res = ToolRunner.run(conf, new AvroReadWriteJob(), args);
            System.exit(res);
        }
    }

最后会将单词的统计结果写入Avro文件中，查看其结果如下：

{"word":"China","count":1}
    {"word":"hello","count":1}
    {"word":"kitty","count":1}
    {"word":"to","count":1}
    {"word":"tom","count":1}
    {"word":"welcome","count":1}
    {"word":"world","count":1}

至此，MapReduce读取Avro的简单示例就以大功告成啦～～

[Avro]: https://download.csdn.net/download/yf_li123/10791043