比较 Spark 和 MapReduce 执行迭代应用Pagerank的性能差异

1. 设计思路

a) MapReduce 执行迭代计算过程中会反复读写 HDFS，因此可以在 HDFS 中观察到每一轮迭代的输出结果。
b) MapReduce 会提交一系列的作业，而 spark 仅有一个应用，在 Yarn 的 UI 显示会不一样。
c) 对于同样规模的数据集，spark 执行时间应当更短。

2. 实验设置

1)Ubuntu18.04、jdk1.8、云主机、IDEA2020.3.4
2) Hadoop2.10.1、Spark2.4.7、Scala2.11.12
3) 数据集：web-google.txt；因为数据集太大了，换成了 mini-web-google.txt，一共有十个结点，并做了一些改动 page.txt
4) 迭代次数：20
5) 阻尼系数：0.85

3.实验过程

1、编写一个 PageRank 应用，观察 HDFS 中的文件。
2、分别基于 MapReduce 和 Spark 编写一个 PageRank 应用，并通过 Yarn 进行提交，观察 Yarn 界面的区别。
3、针对改进的 min-web-google 数据集，分别在 MapReduce 和 Spark 中运行，统计二者的运行时间，并绘制成图表。

4.代码

1、MapReduce
PageRankMapper.java

package cn.edu.ecnu.mapreduce.example.java.pagerank;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;
import java.io.IOException;

// 步骤1：确定输入键值对[K1, V1]的数据类型为[LongWritable, Text]，输出键值对[K2, V2]的数据类型为[Text, ReducePageRankWritable]
public class PageRankMapper extends Mapper {
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        // 步骤2：编写处理逻辑，将[K1, V1]转换为[K2, V2]并输出
        // 以空格为分隔符切割
        String[] pageInfo = value.toString().split(" ");
//        System.out.println(pageInfo);

        // 网页的排名值
        double pageRank = Double.parseDouble(pageInfo[1]);
//        System.out.println(pageRank);

        // 网页的出站链接数
        int outLink = (pageInfo.length-2)/2;
//        System.out.println(outLink);

        ReducePageRankWritable writable;
        writable = new ReducePageRankWritable();
        // 计算贡献值并保存
        writable.setData(String.valueOf(pageRank / outLink));
        // 设置对应的标识
        writable.setTag(ReducePageRankWritable.PR_L); //贡献值

        // 对于每个出战链接，输出贡献值
        for (int i = 2; i < pageInfo.length; i += 2) {
//            System.out.println(pageInfo[i]);
            context.write(new Text(pageInfo[i]), writable);
        }
        writable = new ReducePageRankWritable();

        // 保存网页信息并标识
//        System.out.println(value.toString());
        writable.setData(value.toString());
        writable.setTag(ReducePageRankWritable.PAGE_INFO); //网页信息
        // 以输入的网页信息的网页名称为键进行输出
        context.write(new Text(pageInfo[0]), writable);
    }
}

ReducePageRankWritable.java

package cn.edu.ecnu.mapreduce.example.java.pagerank;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class ReducePageRankWritable implements Writable {

    // 保存网页连接关系(网页信息)或网页排名(贡献值)元组
    private String data;
    // 标识当前对象保存的元组来自网页连接关系还是网页排名
    private String tag;

    // 用于标识的常量
    public static final String PAGE_INFO = "1";
    public static final String PR_L = "2";

    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeUTF(tag);
        dataOutput.writeUTF(data);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        tag = dataInput.readUTF();
        data = dataInput.readUTF();
    }

    // get和set方法
    public String getData() {
        return data;
    }

    public void setData(String data) {
        this.data = data;
    }

    public String getTag() {
        return tag;
    }

    public void setTag(String tag) {
        this.tag = tag;
    }
}

PageRankReducer.java

package cn.edu.ecnu.mapreduce.example.java.pagerank;

import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.io.*;
import javax.print.DocFlavor;
import java.io.IOException;

// 步骤1：确定输入键值对[K2, V2]的数据类型为[Text, ReducePageRankWritable]，确定输出键值对[K3, V3]的数据类型为[Text, NullWritable]
public class PageRankReducer extends Reducer {
    // 阻尼系数
    private static final double D = 0.85;

    @Override
    protected void reduce(Text key, Iterable values, Context context)
        throws IOException, InterruptedException {
        // 步骤2：编写处理逻辑将[K2, V2]转换为[K3, V3]并输出
        String[] pageInfo = null;
        // 从配置项中读取网页的总数
        int totalPage = context.getConfiguration().getInt(PageRank.TOTAL_PAGE, 0);
        // 从配置项中读取网页当前的迭代步数
        int iteration = context.getConfiguration().getInt(PageRank.ITERATION, 0);
        double sum = 0;

        for (ReducePageRankWritable value : values) {
            String tag = value.getTag();
//            System.out.println(tag);

            // 如果是贡献值则进行求和，否则以空格为分隔符切分后保存到pageInfo
            if (tag.equals(ReducePageRankWritable.PR_L)) {
                sum += Double.parseDouble(value.getData());
            }
            else if (tag.equals(ReducePageRankWritable.PAGE_INFO)) {
//                System.out.println(value.getData());
                pageInfo = value.getData().split(" ");
            }
        }
//        System.out.println(sum);

        // 计算排名值
        double pageRank = (1-D) / totalPage + D * sum;
//        System.out.println(pageRank);
        // 跟新网页信息中的排名值
        pageInfo[1] = String.valueOf(pageRank);

        // 最后一次迭代输出网页名以及排名值，而其余迭代输出网页信息
        StringBuilder result = new StringBuilder();
        if (iteration == (PageRank.MAX_ITERATION - 1)) {
            // 保留5位小数
            result.append(pageInfo[0]).append(" ").append(String.format("%.5f", pageRank));
        }
        else {
            for (String data : pageInfo) {
                result.append(data).append(" ");
            }
        }

        context.write(new Text(result.toString()), NullWritable.get());
    }
}

PageRank.java

package cn.edu.ecnu.mapreduce.example.java.pagerank;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
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.*;

public class PageRank extends Configured implements Tool {
    // 设置全局配置项
    public static final int MAX_ITERATION = 20; // 最大迭代步数
    private static int iteration = 0; // 从0开始记录当前迭代步数
    public static final String TOTAL_PAGE = "1"; // 配置项中用于记录网页总数的键
    public static final String ITERATION = "2"; // 配置项中用于记录当前迭代步数的键

    @Override
    public int run(String[] args) throws  Exception {
//        System.out.println(args[0]);
//        System.out.println(args[1]);
//        System.out.println(args[2]);

        // 步骤1：设置作业的信息
//        int totalPage = Integer.parseInt(args[2]);
        int totalPage = 10; 
        getConf().setInt(PageRank.ITERATION, iteration);
        getConf().setInt(PageRank.TOTAL_PAGE, totalPage);

        Job job = Job.getInstance(getConf(), getClass().getSimpleName());
        // 设置程序的类名
        job.setJarByClass(getClass());

        // 设置数据的输入路径
        if (iteration == 0) {
            FileInputFormat.addInputPath(job, new Path(args[0]));
        }
        else {
            // 将上一次迭代的输出设置为输入
            FileInputFormat.addInputPath(job, new Path(args[1] + (iteration - 1)));
        }
        // 设置数据的输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1] + iteration));

        // 设置Map方法及其输出键值对的数据类型
        job.setMapperClass(PageRankMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(ReducePageRankWritable.class);

        // 设置Reduce方法及其输出键值对的数据类型
        job.setReducerClass(PageRankReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        return job.waitForCompletion(true) ? 0 : -1;
    }

    public static void main(String[] args) throws Exception {
        // 步骤2：运行作业，并计算运行时间
        int exitCode = 0;
        long startTime = System.currentTimeMillis();
        while (iteration < MAX_ITERATION) {
            exitCode = ToolRunner.run(new PageRank(), args);
            if (exitCode == -1) {
                break;
            }
            iteration++;
        }
        long endTime = System.currentTimeMillis();
        System.out.println("程序运行时间：" + (endTime - startTime) + "ms");
    }
}

2、Spark
PageRank.scala

package cn.edu.ecnu.spark.example.scala.pagerank

import org.apache.spark.rdd.RDD
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.HashPartitioner
import org.apache.hadoop.fs.FileSystem
import org.apache.hadoop.fs.Path
import java.io.File
import java.util.Date

object PageRank {
  def run(args: Array[String]): Unit = {
    // 步骤1：通过SparkConf设置配置信息，并创建SparkContext
    val conf = new SparkConf
    conf.setAppName("PageRank")
    conf.setMaster("local") // 仅用于本地进行调试，如在集群中运行则删除本行
    val sc = new SparkContext(conf)

    // 步骤2：按应用逻辑使用操作算子编写DAG，其中包括RDD的创建、转换和行动等
    val iterateNum = 20 // 指定迭代次数
    val factor = 0.85 // 阻尼系数

    //    val text = sc.textFile("inputFilePath")
//    System.out.println(args(0))
//    System.out.println(args(1))
    // 读取输入文本数据
    val text = sc.textFile(args(0))

//    System.out.println("==================================")

    // 将文本数据转换成[网页, {链接列表}]键值对
    val links = text.map(line => {
      val tokens = line.split(" ")
      var list = List[String]()
      for (i <- 2 until tokens.size by 2) {
        list = list :+ tokens(i)
      }
      (tokens(0), list)
    }).cache() // 持久化到内存

    // 网页总数
    val N = 10
//    System.out.println(N)

    //初始化每个页面的排名值，得到[网页, 排名值]键值对
    var ranks = text.map(line => {
      val tokens = line.split(" ")
      (tokens(0), tokens(1).toDouble)
    })

    // 执行iterateNum次迭代计算
    for (iter <- 1 to iterateNum) {
      val contributions = links
        // 将links和ranks做join. 得到[网页, {{链接列表}, 排名值}]
        .join(ranks)
        // 计算出每个网页对其每个链接网页的贡献值 = 网页排名值 / 链接总数
        .flatMap {
          case (page, (links, rank)) =>
            links.map(dest => (dest, rank / links.size))
        }

      ranks = contributions
        // 聚合对相同网页的贡献值，求和得到对每个网页的总贡献值
        .reduceByKey(_+_)
        // 根据公式计算得到每个网页的新排名值
        .mapValues(v => (1 - factor) * 1.0 / N + factor * v)
    }

    // 对排名值保留5位小数，并打印最终网页排名结果
    ranks.foreach(t => println(t._1 + " ", t._2.formatted("%.5f")))
    ranks.saveAsTextFile(args(1))

    // 步骤3：关闭SparkContext
    sc.stop()
  }

  def main(args: Array[String]): Unit = {
    // 步骤4：运行作业，并计算运行时间
    var startTime =new Date().getTime
    run(args)
    var endTime =new Date().getTime
    println("程序运行时间：" + (endTime - startTime) + "ms") //单位毫秒
  }
}

3、Input
pagerank.txt

A 1.0 B 1.0 D 1.0
B 1.0 C 1.0
C 1.0 A 1.0 B 1.0
D 1.0 B 1.0 C 1.0

比较 Spark 和 MapReduce 执行迭代应用Pagerank的性能差异

Java相关栏目本月热门文章