logo

小白学苑

让大数据学习更简单

使用KeyValueTextInputFormat

Hadoop支持通过InputFormat来处理多种不同格式和类型的数据。

在Hadoop MapReduce计算中，InputFormat通过解析输入数据生成key-value对作为Mapper的输入。 InputFormat还执行对输入数据的分片，将输入数据切分为逻辑分区，本质上决定了MapReduce计算的Map任务数量，并间接地决定了这些Map任务的执行位置。 Hadoop为每个逻辑数据分区生成一个Map任务，并调用相应的Mapper，将逻辑切分的key-value对作为输入。

确保Mapper的输入数据类型与MapReduce计算所使用的InputFormat生成的数据类型相匹配。

Hadoop内置InputFormat实现

下面是Hadoop提供的一些InputFormat实现，以支持许多常见的数据格式：

  • TextInputFormat: 这用于普通文本文件。TextInputFormat会为输入的文本文件的每一行都生成一个key-value record。对于每一行，key(LongWritable)是该行在文件中的字节偏移量，value(Text)是该行的文本。TextInputFormat是Hadoop默认的InputFormat。
  • KeyValueTextInputFormat：这是一个用于普通文本文件的输入格式，它会为输入的文本文件的每一行生成一个key-value记录。输入数据的每一行按照分隔字符被拆分为一个key(text)和value(text)对。默认的分隔符是tab字符。如果一行不包含该分隔符的话，则整行将被当作key，而value将会为空。
  • NLineInputFormat: 这用于普通 的文本文件。NLineInputFormat将输入文件拆分为固定数量行的逻辑分区。如果想让Map任务接收固定数量的行作为输入的话，可以使用NLineInputFormat。其中key(LongWritable)和value(Text)是split中的每一行生成的，与TextInputFormat类相似。默认地，NLineInputFormat为每行创建一个逻辑split(以及一个Map任务)。每个split的行的数量(或每个Map任务的key-value记录)可以像下面这样指定。NLineInputFormat为输入文本文件的每一行生成一个key-value record： 
        NLineInputFormat.setNumLinesPerSplit(job,50);
  • SequenceFileInputFormat: 这用于Hadoop SequenceFile输入数据。Hadoop SequenceFile将数据作为二进制的key-value对进行存储，并支持数据压缩。当使用前一个MapReduce计算的SequenceFile格式结果作为后一个MapReduce计算的输入时，使用SequenceFileInputFormat。下面是其子类：
    • SequenceFileAsBinaryInputFormat: 这是SequenceInputFormat的子类，以raw binary格式表示key(BytesWritable)和value(BytesWritable)对。
    • SequenceFileAsTextInputFormat: 这是SequenceInputFormat的子类，将key(Text)和value(Text)对表示为字符串。
  • DBInputFormat：支持从一个SQL表读取数据作为MapReduce计算的输入。DBInputFormat使用记录号作为key(LongWritable)，以查询结果记录作为value(DBWritable)。

了解KeyValueTextInputFormat

KeyValueTextInputFormat是一个用于普通文本文件的输入格式，它会为输入的文本文件的每一行生成一个key-value记录。 输入数据的每一行按照分隔字符被拆分为一个key(text)和value(text)对。默认的分隔符是tab字符。 如果一行不包含该分隔符的话，则整行将被当作key，而value将会为空。

我们可以通过在job的配置对象中设置一个属性来指定一个自定义的分隔，如下所示，使用逗号字符作为key和value的分隔符：

    conf.set("key.value.separator.in.input.line", ",");

KeyValueTextInputFormat是基于FileInputFormat, FileInputFormat是用于基于文件的InputFormats的基类。 因此，使用FileInputFormat类的setInputPaths()方法一指定MapReduce计算的输入路径。 当使用任何基于FileInputFormat类的InputFormat时，必须执行下面这一步：

    FileInputFormat.setInputPaths(job, new Path(inputPath));

可以向MapReduce计算提供多个HDFS输入路径，路径列表之间用逗号分隔。 还可以使用FileInputFormat类的静态方法addInputPath()向计算添加额外的输入路径：

    public static void setInputPaths(JobConf conf,Path… inputPaths)
    public static void addInputPath(JobConf conf, Path path)

下面的演示了怎样使用基于FileInputFormat的KeyValueTextInputFormat作为用于Hadoop MapReduce计算的InputFormat。

一、创建Java Maven项目

Maven依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>HadoopDemo</groupId>
    <artifactId>com.xueai8</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <!--hadoop依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--hdfs文件系统依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--MapReduce相关的依赖-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
            <version>3.3.1</version>
        </dependency>
        <!--junit依赖-->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!--编译器插件用于编译拓扑-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <!--指定maven编译的jdk版本和字符集,如果不指定,maven3默认用jdk 1.5 maven2默认用jdk1.3-->
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source> <!-- 源代码使用的JDK版本 -->
                    <target>1.8</target> <!-- 需要生成的目标class文件的编译版本 -->
                    <encoding>UTF-8</encoding><!-- 字符集编码 -->
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

KvInputFormatDemo.java：

因为代码比较简单，我们把mapper和reducer作为静态嵌套类。

package com.xueai8.logkvinput;

import java.io.IOException;

import com.xueai8.job512.Job51Mapper2;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * 输出数据，每行用Tab分隔。KeyValueTextInputFormat会自动进行分割，分别作为key和value传入map函数。
 * a	hello
 * b 	hadoop
 */
public class KvInputFormatDemo {

    public static class KvMapper extends Mapper<Text, Text, Text, Text> {

        Logger logger = LoggerFactory.getLogger(KvInputFormatDemo.class);

        // KeyValueTextInputFormat会自动按Tab进行分割，组织成key-value传入map函数
        @Override
        protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
            logger.info("=== key: " + key + ", value: " + value);
            context.write(key, value);
        }
    }

    public static class KvReducer extends Reducer<Text, Text, Text, Text> {

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            for (Text value : values) {
                context.write(value, key);
            }
        }

    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        if (args.length < 2) {
            System.err.println("语法:  <input_path> <output_path>");
            System.exit(-1);
        }

        Configuration conf = new Configuration();
//        conf.set("key.value.separator.in.input.line","\t");    // 可以指定分隔符

        Job job = Job.getInstance(conf, "kvinput format test");
        job.setJarByClass(KvInputFormatDemo.class);

        job.setMapperClass(KvMapper.class);
        job.setReducerClass(KvReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setInputFormatClass(KeyValueTextInputFormat.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

二、配置log4j

在src/main/resources目录下新增log4j的配置文件log4j.properties，内容如下：

log4j.rootLogger = info,stdout

### 输出信息到控制抬 ###
log4j.appender.stdout = org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target = System.out
log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

三、项目打包

打开IDEA下方的终端窗口terminal，执行"mvn clean package"打包命令，如下图所示：

如果一切正常，会提示打jar包成功。如下图所示：

这时查看项目结构，会看到多了一个target目录，打好的jar包就位于此目录下。如下图所示：

四、项目部署

请按以下步骤执行。

1、启动HDFS集群和YARN集群。在Linux终端窗口中，执行如下的脚本：

$ start-dfs.sh
$ start-yarn.sh

查看进程是否启动，集群运行是否正常。在Linux终端窗口中，执行如下的命令：

$ jps

这时应该能看到有如下5个进程正在运行，说明集群运行正常：

    5542 NodeManager
    5191 SecondaryNameNode
    4857 NameNode
    5418 ResourceManager
    4975 DataNode

2、在本地先编辑一个数据文件input.txt，内容如下：

a	hadoop
b	spark

然后将该数据文件input.txt上传到HDFS的/data/mr/目录下。

$ hdfs dfs -mkdir -p /data/mr
$ hdfs dfs -put input.txt /data/mr/
$ hdfs dfs -ls /data/mr/

3、提交作业到Hadoop集群上运行。（如果jar包在Windows下，请先拷贝到Linux中。）

在终端窗口中，执行如下的作业提交命令：

$ hadoop jar HadoopDemo-1.0-SNAPSHOT.jar com.xueai8.logkvinput.KvInputFormatDemo /data/mr /data/mr-output

4、查看输出结果。

在终端窗口中，执行如下的HDFS命令，查看输出结果：

$ hdfs dfs -ls /data/mr-output 
$ hdfs dfs -cat /data/mr-output/part-r-00000

可以看到最后的统计结果如下：

hadoop	a
spark	b
《PySpark原理深入与编程实战》