mapReduce清除日志中字 段长度比 11 小的日志记录

Question

打开 ZZ052-7-M2-T3-SUBT2 文件夹，文件夹中包含
sku_info.csv 文件。sku_info.csv 文件存储了电商互联网
平台上收集的商品数据，数据中有以下内容：
id：主键非空，bigint 类型，长度为 20
spu_id：spuId，varchar 类型，长度 20
price：价格，decimal 类型，长度 10
sku_name：商品名称，varchar 类型，长度 200
sku_desc：商品描述，varchar 类型，长度 2000
weight:重量，decimal 类型，长度 10
tm_id:品牌，bigint 类型，长度 20
category3_id:三级分类，bigint 类型，长度 20
sku_default_img:默认显示图片，varchar 类型，长度
200
编写 MapReduce 程序，实现以下功能：清除日志中字
段长度比 11 小的日志记录，输出文件到 HDFS；在控制台按
顺序打印输出前 20 条数据

Answer 1

引自免费微信小程序：皆我百晓生

为了实现这个MapReduce程序，你需要首先安装Hadoop并设置好环境变量。然后按照以下步骤操作：

1. 创建Java项目
2. 添加Hadoop依赖
3. 编写Mapper类
4. 编写Reducer类
5. 编写Driver类
6. 执行程序

以下是详细的步骤和代码示例。

步骤1：创建Java项目

创建一个新的Java Maven项目，并命名为logClearing.

步骤2：添加Hadoop依赖

在pom.xml文件中添加Hadoop的依赖。如果你使用的是Hadoop 3.x版本，可以使用以下依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>3.3.1</version>
    </dependency>
</dependencies>

步骤3：编写Mapper类

创建一个名为LogLengthMapper.java的文件，实现Mapper<LongWritable, Text, Text, IntWritable>接口。将输入的日志行分割成单词，计算每个单词的长度，只将长度大于等于11的单词输出。

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 LogLengthMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] words = line.split("\\s+");
        for (String word : words) {
            if (word.length() >= 11) {
                this.word.set(word);
                context.write(this.word, one);
            }
        }
    }
}

步骤4：编写Reducer类

创建一个名为LogLengthReducer.java的文件，实现Reducer<Text, IntWritable, Text, IntWritable>接口。将相同单词的计数累加起来。

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class LogLengthReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        result.set(sum);
        context.write(key, result);
    }
}

步骤5：编写Driver类

创建一个名为LogLengthDriver.java的文件，定义主入口点。

import org.apache.hadoop.conf.Configuration;
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;

public class LogLengthDriver {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Log Length Clearer");
        job.setJarByClass(LogLengthDriver.class);
        job.setMapperClass(LogLengthMapper.class);
        job.setCombinerClass(LogLengthReducer.class);
        job.setReducerClass(LogLengthReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

步骤6：执行程序

在命令行中运行以下命令，替换input_path和output_path为实际路径：

mvn compile exec:exec -Dexec.mainClass="LogLengthDriver" -Dexec.args="<input_path> <output_path>"

这将会读取指定的输入目录（input_path），执行MapReduce任务，清理长度小于11的单词，并将结果保存到指定的输出目录（output_path）。

注意：此程序不会在控制台上打印前20条数据，因为MapReduce是分布式处理，数据在节点间传输，不支持直接在控制台查看中间结果。如果你想在控制台上查看结果，你可能需要修改Reducer类，在完成所有reduce操作后，手动写入控制台。但这样会破坏MapReduce的并行性和效率，仅适用于测试和调试目的。