使用Java跨平台实现Mapreduce词频统计

本文将演示通过JavaAPI在Hadoop集群上使用Mapreduce进行词频统计的简易过程：

创建Maven项目并本地编写Java代码
将要处理的数据上传至Hadoop系统
将Maven项目打包成jar包并上传至服务器运行

系统：Win10，CentOS 7.6

软件：Maven 3.8.4，Hadoop 2.7.3，Xshell，Xftp

1.创建Maven项目并本地编写Java代码

初始化Maven项目

首先创建一个新的Maven项目，此时文件结构应该类似于：

使用Java跨平台实现Mapreduce词频统计

此时将

pom.xml

文件的内容全部替换成如下：

<?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>org.example</groupId>
    <artifactId>hdfs_upload</artifactId>
    <version>1.0-SNAPSHOT</version>

    <name>hdfs_upload</name>
    <!-- FIXME change it to the project's website -->
    <url>http:///maven.apache.org</url>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.7</maven.compiler.source>
        <maven.compiler.target>1.7</maven.compiler.target>
    </properties>

    <dependencies>
        <!-- hadoop基础库 -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>3.1.3</version>
        </dependency>
        <!-- hadoop客户端 -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.3</version>
        </dependency>
        <!-- hadoop hdfs -->
        <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-hdfs -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>3.1.3</version>
            <scope>test</scope>
        </dependency>
        <!-- hadoop-mapreduce -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-mapreduce-client-core</artifactId>
            <version>3.1.3</version>
        </dependency>

        <!-- 单元测试 -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
        </dependency>
        <!-- 控制⽇志信息输出 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.25</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

如果有报错，有某个包找不到，就在终端输入：
mvn -U idea:idea
           
更新或者继续下载未安装完的包，稍等下载至报错信息消失就可以了！

使用

pom.xml

文件重新构建Maven项目，点击

重新加载项目

使用Java跨平台实现Mapreduce词频统计

本地编写java代码

以最简单的词频统计为例，我们对word.txt文件进行每个单词出现次数的统计

word.txt文件内容如下：

doop hbase
hive hdfs
hadoop zookeeper
hdfs mapreduce
yarn hadoop

在项目目录下

src/main/java

下创建org.example文件夹

在项目目录下

src/main/java/org.example

下创建

WordCountMain
WordCountMapper
WordCountReducer

三个java文件，并分别编写：

WordCountMain

package org.example;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 java.io.IOException;
import java.net.URISyntaxException;
public class WordCountMain {
    public static void main(String[] args) throws IOException,
            InterruptedException, ClassNotFoundException, URISyntaxException {
        // 创建hadoop集群配置对象
        Configuration conf = new Configuration();
        // 创建⼀个 job 实例
        Job job = Job.getInstance(conf,"word count");
        // 设置主类
        job.setJarByClass(WordCountMain.class);

        // 设置 job 的 mapper 类
        job.setMapperClass(WordCountMapper.class);
        // 设置 Mapper 的输出键类型
        job.setMapOutputKeyClass(Text.class);
        // 设置 Mapper 的输出值类型
        job.setMapOutputValueClass(IntWritable.class);

        // 设置 job 的 reducer 类
        job.setReducerClass(WordCountReducer.class);
        // 设置 Reducer 的输出键类型
        job.setOutputKeyClass(Text.class);
        // 设置 Reducer 的输出值类型
        job.setOutputValueClass(IntWritable.class);
        // 指定 job 的输⼊⽂件路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        // 指定 job 的输出⽂件路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        Path path = new Path(args[1]);
        FileSystem fs = FileSystem.get(conf);
        // 判断⽬录是否存在，如果存在，删除该⽬录
        if (fs.exists(path)) {
            fs.delete(path,true);
        }
        // 等待任务结束
        System.exit(job.waitForCompletion(true)?0:1);
    }
}

WordCountMapper

package org.example;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;
public class WordCountMapper extends Mapper<LongWritable, Text, Text,IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        Text k = new Text();
        IntWritable v = new IntWritable(1);
        // 将Text类型⽂本转换为String类型
        String text = value.toString();
        // 分词，存储到String数组
        String[] words = text.split(" ");
        // 输出
        for (String word : words) {
            // k.set(word)：将word装载到k中
            k.set(word);
            // 将map()函数输出的键值对写⼊到MapReduce上下⽂环境
            context.write(k, v);
        }
    }
}

WordCountReducer

package org.example;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class WordCountReducer extends Reducer<Text, IntWritable, Text,
        IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,
                          Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws
            IOException, InterruptedException {
        int sum=0;
        IntWritable v = new IntWritable();
        for (IntWritable count : values) {
            // count.get()：获取 count 值
            sum += count.get();
        }
        // 将sum值装载到v中
        v.set(sum);
        // 将reduce()函数输出的键值对写⼊到MapReduce上下⽂环境
        context.write(key,v);
    }
}

无报错就说明没问题了！

配置日志文件

在

src/main/resources

目录下创建文件

log4j.properties

，并在复制以下文本到其中：

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

2.将要处理的数据上传至Hadoop系统

本文将要处理的数据名为：

word.txt

windows->linux

使用Xftp将文件传至Linux系统中的

/opt

目录下

linux->hadoop

首先关闭防火墙，然后

start-all.sh

开启hadoop，能够打开hadoop的Web端则为成功

先在hadoop中创建一个目录

hadoop fs -mkdir /mapreduce

将数据上传至其中

hadoop fs -put /opt/word.txt /mapreduce/word.txt

然后就可以在hadoop的Web端中看到数据文件：

使用Java跨平台实现Mapreduce词频统计

3.将Maven项目打包成jar包并上传至服务器运行

点击右侧栏中Maven一项，双击package，对整个Maven项目进行打包：

使用Java跨平台实现Mapreduce词频统计

运行结果如下：

使用Java跨平台实现Mapreduce词频统计

在结果展示的位置中可以找到jar包：

使用Java跨平台实现Mapreduce词频统计

将

original-hdfs_upload-1.0-SNAPSHOT

通过Xftp传至服务器的

/opt

目录下

运行jar包

hadoop jar original-hdfs_upload-1.0-SNAPSHOT.jar org.example.WordCountMain /mapreduce/word.txt /mapreduce/output

其中

org.example.WordCountMain

为将要运行的类名

/mapreduce/word.txt

为输入文件在hadoop中的位置

/mapreduce/output

为输出文件在hadoop中存放的文件夹位置

接下来会输出一长串hadoop的运行过程，在其中可以找到以下内容：

使用Java跨平台实现Mapreduce词频统计

则为运行成功！

接下来我们可以在Web端找到输出结果的文件，并下载：

使用Java跨平台实现Mapreduce词频统计

将

part-r-00000

改为

part-r-00000.txt

就可以查看内容了：

使用Java跨平台实现Mapreduce词频统计

使用Java跨平台实现Mapreduce词频统计

1.创建Maven项目并本地编写Java代码

初始化Maven项目

本地编写java代码

WordCountMain

WordCountMapper

WordCountReducer

配置日志文件

2.将要处理的数据上传至Hadoop系统

windows->linux

linux->hadoop

3.将Maven项目打包成jar包并上传至服务器运行

继续阅读

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

Ambari介绍和架构原理

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

maven No compiler is provided in this environment. Perhaps you are running on a JRE rather than a J

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method

Opendaylight课堂之深度剖析toaster（一）