Spark007--综合案例

1.数据如下
    a,1,3
    a,2,4
    b,1,1
根据第一列统计出
    a,3,7
    b,1,1
用RDD实现

分析：
1）使用逗号对数据进行拆分 (a,<1,3>)  a=_.1  <1,3>=_.2
2）reduceByKey((a,b)=>a+b)   =>  _.2._1 + _.2._2

package com.ruozedata.spark.spark04
import com.ruozedata.spark.homework.utils.ContextUtils
import com.ruozedata.spark.homework.utils.ImplicitAspect._
object InterviewApp01 {
  def main(args: Array[String]): Unit = {
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
    val input = sc.parallelize(List(
      List("a",1,3),
      List("a",2,4),
      List("b",1,1)
    ))
    input.map(x => {
      val key = x(0).toString
      val v1 = x(1).toString.toInt
      val v2 = x(2).toString.toInt
      (key, (v1,v2))
    }).reduceByKey((x,y)=>{
      (x._1 + y._1, x._2+y._2)
    }).map(x=>List(x._1, x._2._1,x._2._2)).printInfo()
    sc.stop()
  }
}

结果是：
List(a, 3, 7)
List(b, 1, 1)
-------------------------

2.广告投放 收费标准：
	看到就收费
	点击 才收费
eg：
"1000000,一起看|电视剧|军旅|士兵突击,1,0"    
 uid,导航,1,0       （1表示看到了 0表示没有点进去）

需求1：人和“一个东西”的展示量以及点击量
eg：1000000 一起看 2 1

使用reduceBykey和groupBykey都实现一下：

package com.ruozedata.spark.spark04

import com.ruozedata.spark.homework.utils.ContextUtils
import com.ruozedata.spark.homework.utils.ImplicitAspect._
object InterviewApp02 {
    def main(args: Array[String]): Unit = {
      val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
      val input = sc.parallelize(List(
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,1,0 1表示看到了 0表示没有点进去
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1"
      ))

      /**
        * 需求：人和“一个东西”的展示量以及点击量
        * 1）组合key：人和所谓的一个东西
        *
        * 1000000 一起看 2 1
        */
      val processRDD = input.flatMap(x => {
        val splits = x.split(",")
        val id = splits(0).toInt
        val word = splits(1)
        val show = splits(2).toInt
        val clicks = splits(3).toInt

        val words = word.split("\\|")
        words.map(x => ((id, x), (show, clicks)))
      })

      /**
        * 在每个task/partition按照key先进行一个本地的聚合mapSideCombine: Boolean = true
        * 预聚合之后，在每个task之上对于相同key的数据只有一条
        *
        *
        * 调优 前 vs 后
        * 是否能按照调优之前和调优之后作业的执行时间来对比?
        * 时间之外还有其他的：读进来多少数据，shuffle出去多少数据，shuffle读写花费多少时间
        * 
        * 所以优先选择reduceByKey 
        * 	1.shuffle数据少
        * 去4040页面查看就知道了
        */
      processRDD.reduceByKey((x,y)=>(x._1+y._1, x._2+y._2)).printInfo()

      // 数据全部进行shuffle操作
      processRDD
        .groupByKey().mapValues(x=>{
        val totalShows = x.map(_._1).sum
        val totalClicks = x.map(_._2).sum
        (totalShows, totalClicks)
      }).printInfo()

      sc.stop()
    }

}
结果是：
((1000000,军旅),(10,5))
((1000000,一起看),(10,5))
((1000000,士兵突击),(10,5))
((1000001,电视剧),(5,5))
((1000001,我的团长我的团),(5,5))
((1000001,一起看),(5,5))
((1000001,军旅),(5,5))
((1000000,电视剧),(10,5))
-------------------------
((1000000,一起看),(10,5))
((1000000,军旅),(10,5))
((1000001,电视剧),(5,5))
((1000000,士兵突击),(10,5))
((1000001,军旅),(5,5))
((1000001,我的团长我的团),(5,5))
((1000001,一起看),(5,5))
((1000000,电视剧),(10,5))
-------------------------

在这里插入图片描述

一定要注意数据结构

object InterviewApp02 {
    def main(args: Array[String]): Unit = {
      val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
      val input = sc.parallelize(List(
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,1,0 1表示看到了 0表示没有点进去
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1",
        "1000000,一起看|电视剧|军旅|士兵突击,1,0", // uid,导航,
        "1000000,一起看|电视剧|军旅|士兵突击,1,1",
        "1000001,一起看|电视剧|军旅|我的团长我的团,1,1"
      ))

      /**
        * 需求：人和“一个东西”的展示量以及点击量
        * 1）组合key：人和所谓的一个东西
        *
        * 1000000 一起看 2 1
        * flatMap 注意 为什么不用map *** 
        */
      val processRDD = input.flatMap(x => {
        val splits = x.split(",")
        val id = splits(0).toInt
        val word = splits(1)
        val show = splits(2).toInt
        val clicks = splits(3).toInt

        val words = word.split("\\|")
        words.map(x => ((id, x), (show, clicks)))
      })
      // 数据全部进行shuffle操作
      processRDD
        .groupByKey().printInfo()
      sc.stop()
    }
}

结果是：
((1000000,一起看),CompactBuffer((1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1)))
((1000000,军旅),CompactBuffer((1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1)))
((1000001,电视剧),CompactBuffer((1,1), (1,1), (1,1), (1,1), (1,1)))
((1000000,士兵突击),CompactBuffer((1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1)))
((1000001,军旅),CompactBuffer((1,1), (1,1), (1,1), (1,1), (1,1)))
((1000001,我的团长我的团),CompactBuffer((1,1), (1,1), (1,1), (1,1), (1,1)))
((1000000,电视剧),CompactBuffer((1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1), (1,0), (1,1)))
((1000001,一起看),CompactBuffer((1,1), (1,1), (1,1), (1,1), (1,1)))


groupByKey 没有预聚合 还记得wc那张图么 上一篇的。

3.  分组排序/组内排序
  求每个域名访问量最大的url的Top N

数据格式：
www.baidu.com,url1
www.baidu.com,url2
www.baidu.com,url2
www.baidu.com,url3
www.baidu.com,url3
www.baidu.com,url3
www.baidu.com,url4

我们一步步来：

object InterviewApp03 {

  def main(args: Array[String]): Unit = {

    val TOPN = 2
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/site.log")
    val processRDD = input.map(x => {
      val splits = x.split(",")
      val site = splits(0)
      val url = splits(1)
      ((site, url), 1)
    })

    processRDD.printInfo()

    sc.stop()
  }

}

结果是：
((www.google.com,url6),1)
((www.baidu.com,url1),1)
((www.baidu.com,url2),1)
((www.google.com,url6),1)
((www.baidu.com,url2),1)
((www.google.com,url2),1)
。。。。。
-------------------------

分组：如何分组？不分组行不行？

object InterviewApp03 {

  def main(args: Array[String]): Unit = {

    val TOPN = 2
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/site.log")
    val processRDD = input.map(x => {
      val splits = x.split(",")
      val site = splits(0)
      val url = splits(1)
      ((site, url), 1)
    })

        processRDD.reduceByKey(_+_)
          .groupBy(_._1._1)
          .printInfo()
    sc.stop()
  }
}

结果是：
(www.baidu.com,CompactBuffer(
((www.baidu.com,url1),1), 
((www.baidu.com,url3),3),
 ((www.baidu.com,url5),5),
  ((www.baidu.com,url2),2),
   ((www.baidu.com,url4),4)))
   
(www.twitter.com,CompactBuffer(((www.twitter.com,url9),6), ((www.twitter.com,url10),11), ((www.twitter.com,url6),1)))
(www.google.com,CompactBuffer(((www.google.com,url6),7), ((www.google.com,url8),7), ((www.google.com,url1),1), ((www.google.com,url2),2)))
-------------------------

结果：

object InterviewApp03 {

  def main(args: Array[String]): Unit = {

    val TOPN = 2
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/site.log")
    val processRDD = input.map(x => {
      val splits = x.split(",")
      val site = splits(0)
      val url = splits(1)
      ((site, url), 1)
    })
       processRDD.reduceByKey(_+_)
          .groupBy(_._1._1)
            .mapValues(x => {
              x.toList.sortBy(-_._2)  // toList是一个很大的安全隐患    
                .map(x => (x._1._2, x._2)).take(TOPN)
            }).printInfo()
    sc.stop()
  }
}

结果是：
(www.baidu.com,List((url5,5), (url4,4)))
(www.twitter.com,List((url10,11), (url9,6)))
(www.google.com,List((url6,7), (url8,7)))
-------------------------

x.toList.sortBy(-.2) // toList是一个很大的安全隐患，为什么这么说呢？
x来了一亿条数据 list就炸掉了所以这样虽然能出结果但是不能用
如何规避掉呢这块就使用rdd算子不用scala的高级函数

如何解决呢？
1.方法：
	分而治之的思路
		类似mapreduce 思想  一个文件 拆成多个inputsplits 每个split单独处理 之后reduce聚合

object InterviewApp03 {
  def main(args: Array[String]): Unit = {
    val TOPN = 2
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/site.log")
    val processRDD = input.map(x => {
      val splits = x.split(",")
      val site = splits(0)
      val url = splits(1)
      ((site, url), 1)
    })
    // 分而治之的思路
        val sites = Array("www.baidu.com","www.google.com","www.twitter.com")
        for(site <- sites) {
          processRDD.filter(_._1._1 == site)
            .reduceByKey(_+_).sortBy(-_._2)
            .take(TOPN).foreach(println)
        }
    sc.stop()
  }
}

结果是：
((www.baidu.com,url5),5)
((www.baidu.com,url4),4)
((www.google.com,url6),7)
((www.google.com,url8),7)
((www.twitter.com,url10),11)
((www.twitter.com,url9),6)

有什么问题？
1.会产生好多job  去ui上看 
2.val sites = Array("www.baidu.com","www.google.com","www.twitter.com") 不要这么写

方法2; 优化方法1

object InterviewApp03 {

  def main(args: Array[String]): Unit = {

    val TOPN = 2
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/site.log")
    val processRDD = input.map(x => {
      val splits = x.split(",")
      val site = splits(0)
      val url = splits(1)
      ((site, url), 1)
    })
    // 分而治之的思路
    val sites = processRDD.map(_._1._1).distinct().collect()  // 数组
    sites.map(x=>{
      processRDD.filter(_._1._1 == x).reduceByKey(_+_).sortBy(-_._2) .take(TOPN).foreach(println)
    })
    sc.stop()
  }
}

结果是：
((www.baidu.com,url5),5)
((www.baidu.com,url4),4)
((www.twitter.com,url10),11)
((www.twitter.com,url9),6)
((www.google.com,url6),7)
((www.google.com,url8),7)

问题：
 val sites = processRDD.map(_._1._1).distinct().collect()  // 数组
生产上能直接collect么？不能 这代码怎么改进呢？

Partitioner

 def reduceByKey(func: (V, V) => V): RDD[(K, V)] = self.withScope {
    reduceByKey(defaultPartitioner(self), func)
  }

reduceByKey走的是 defaultPartitioner


def defaultPartitioner(rdd: RDD[_], others: RDD[_]*): Partitioner = {
    val rdds = (Seq(rdd) ++ others)
    val hasPartitioner = rdds.filter(_.partitioner.exists(_.numPartitions > 0))
    if (hasPartitioner.nonEmpty) {
      hasPartitioner.maxBy(_.partitions.length).partitioner.get
    } else {
      if (rdd.context.conf.contains("spark.default.parallelism")) {
        new HashPartitioner(rdd.context.defaultParallelism)
      } else {
        new HashPartitioner(rdds.map(_.partitions.length).max)
      }
    }
  }

RDD的5大特性中有一条是 Partitioner

在这里插入图片描述

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)
      : RDD[(K, V)] = self.withScope
  {
    val part = new RangePartitioner(numPartitions, self, ascending)
    new ShuffledRDD[K, V, V](self, part)
      .setKeyOrdering(if (ascending) ordering else ordering.reverse)
  }

RangePartitioner：

object PartitionerApp {

  def main(args: Array[String]): Unit = {
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
  
    val data = sc.parallelize(List(1,2,3,4,5,6,30,100,300,400,500),3)
  //Kafka分区策略
    data.zipWithIndex().sortByKey()
      .mapPartitionsWithIndex((index, partition)=>{
        partition.map(x=>s"分区是$index, 元素是${x._1}")
      }).printInfo()

    sc.stop()
  }
}

结果是：
分区是0, 元素是1
分区是0, 元素是2
分区是0, 元素是3
分区是0, 元素是4
分区是1, 元素是5
分区是1, 元素是6
分区是1, 元素是30
分区是1, 元素是100
分区是2, 元素是300
分区是2, 元素是400
分区是2, 元素是500
-------------------------

def groupByKey(): RDD[(K, Iterable[V])] = self.withScope {
    groupByKey(defaultPartitioner(self))
  }

defaultPartitioner --》HashPartitioner

object PartitionerApp {
  def main(args: Array[String]): Unit = {
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)
   
    val data = sc.parallelize(List(1,2,3,4,5,6,30,100,300,400,500),3)
    data.zipWithIndex().groupByKey()
      .mapPartitionsWithIndex((index, partition)=>{
        partition.map(x=>s"分区是$index, 元素是${x._1}")
      }).printInfo()

    sc.stop()
  }
}

结果是;
分区是0, 元素是300
分区是0, 元素是30
分区是0, 元素是6
分区是0, 元素是3
分区是1, 元素是100
分区是1, 元素是4
分区是1, 元素是1
分区是1, 元素是400
分区是2, 元素是500
分区是2, 元素是5
分区是2, 元素是2
-------------------------

多路径输出 ***定制化业务

日志格式：
uid1	Andriod	v3	215.197.96.120	4780	2019-09-11 08:37:33	9232	app19-20-18	香港			2019	09	11
uid4	Symbain	71.10.97	168.170.39.193	189	2019-09-11 01:44:36	9232	app19-20-18	湖北	武汉	联通	2019	09	11
uid2	linux	v2	168.170.39.193	189	2019-09-11 03:08:19	9232	app15-1-14-11	湖北	武汉	联通	2019	09	11
uid3	linux	v2	168.170.39.193	189	2019-09-11 03:51:14	9232	app15-1-14-11	湖北	武汉	联通	2019	09	11
uid6	mac	71.10.97	171.125.131.128	4780	2019-09-11 06:54:29	9232	app19-20-18	山西	忻州	联通	2019	09	11
uid10	mac	71.10.97	171.125.131.128	4780	2019-09-11 07:09:27	189	app15-1-14-11	山西	忻州	联通	2019	09	11

eg：给你一个完整的日志 做成 客户定制版  （可以去各大云平台cdn上查看产品）
这个功能很重要 定制是收钱的（这个功能 公司一年收益是很多的 ）

       1.假设按照不同的 品牌进行落盘
       Andriod的数据都落在 Andriod的文件夹下
       Symbain的数据都落在 Symbain的文件夹下
      2. 输出的日志 客户想要什么字段的日志就输出什么字段的日志 （而不是把全部的日志都输出）

1.假设按照不同的品牌进行落盘

输出的时候相当于根据某一个字段进行输出

我们一步一步的来：

object MulitOutputApp {

  def main(args: Array[String]): Unit = {
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)

    val output = "file:///C:/IdeaProjects/spark/out/mulit"
    /**
      * Android
      *   xxxx.log
      *
      * iOS
      *   xxx.log
      */
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/access.log")
    input.map(x => {
      val splits = x.split("\t")
      (splits(1), x) // (platform , 完整的日志)
    }).partitionBy(new HashPartitioner(8))
      .saveAsTextFile(output)
    sc.stop()
  }
}

这个结果肯定是不对的 都放到一个目录下的

在这里插入图片描述

所有日志到在一个文件夹下：
part-00000：mac、linux、Symbain  的 数据
part-00001：Andriod 
part-00002 ：空
part-00003：空
part-00004：windows
	
	不仅仅是日志到在一个文件夹下 而且有的同一个文件下有别的品牌的数据

1.我们是要把某一个字段作为输出文件夹名 
这就是多目录输出 
2.mapreduce里面有这个类  MultipleTextOutputFormat
所以我们自己实现一个类继承这个类就可以

/**
 * This class extends the MultipleOutputFormat, allowing to write the output
 * data to different output files in Text output format.
 */
@InterfaceAudience.Public
@InterfaceStability.Stable
public class MultipleTextOutputFormat<K, V>
    extends MultipleOutputFormat<K, V> {

  private TextOutputFormat<K, V> theTextOutputFormat = null;

  @Override
  protected RecordWriter<K, V> getBaseRecordWriter(FileSystem fs, JobConf job,
      String name, Progressable arg3) throws IOException {
    if (theTextOutputFormat == null) {
      theTextOutputFormat = new TextOutputFormat<K, V>();
    }
    return theTextOutputFormat.getRecordWriter(fs, job, name, arg3);
  }
}

class MyDataMultipleTextOutputFormat extends MultipleTextOutputFormat[Any,Any]{
    补充里面的实现方法
  }

在这里插入图片描述
没有自己想要的方法，继续看MultipleTextOutputFormat的父类

 class MyDataMultipleTextOutputFormat extends MultipleTextOutputFormat[Any,Any]{
    override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = {
      s"$key/$name"   
    }

  }

注意：
 1.
   s"$key/$name"   
   key和name是什么东西呢？一会debug看一下
2.这个东西写好了之后怎么用呢？
 第一个测试代码基础上 就不能使用saveAsTextFile
 因为你要定向输出到某一个类里面去 需要设置FileOutputFormat

这和在mapreduce里是一样的 
这块就要使用 saveAsHadoopFile

在这里插入图片描述

/**
 * Output the RDD to any Hadoop-supported file system, using a Hadoop `OutputFormat` class
 * supporting the key and value types K and V in this RDD. Compress with the supplied codec.
 */
def saveAsHadoopFile(
    path: String,
    keyClass: Class[_],
    valueClass: Class[_],
    outputFormatClass: Class[_ <: OutputFormat[_, _]],
    codec: Class[_ <: CompressionCodec]): Unit = self.withScope {
  saveAsHadoopFile(path, keyClass, valueClass, outputFormatClass,
    new JobConf(self.context.hadoopConfiguration), Some(codec))
}

object MulitOutputApp {

  def main(args: Array[String]): Unit = {
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)

    val output = "file:///C:/IdeaProjects/spark/out/mulit"
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/access.log")
    input.map(x => {
      val splits = x.split("\t")
      (splits(1), x) // (platform , 完整的日志)
    }).partitionBy(new HashPartitioner(5))
      .saveAsHadoopFile(output,classOf[String],classOf[String],classOf[RuozedataMultipleTextOutputFormat])
    sc.stop()
  }

  class RuozedataMultipleTextOutputFormat extends MultipleTextOutputFormat[Any,Any]{
    override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = {
      s"$key/$name"
    }
  }
}

查看结果：

在这里插入图片描述

怎么才能去掉呢？

知道了kv代表了什么。那么如何去掉文件里的多出那一列key值呢？

在这里插入图片描述

默认generateActualKey 返回是最终输出的key  所以我们自定义的类里 重写这个方法 就ok了
  /**
   * Generate the actual key from the given key/value. The default behavior is that
   * the actual key is equal to the given key
   * 
   * @param key
   *          the key of the output data
   * @param value
   *          the value of the output data
   * @return the actual key derived from the given key/value
   */
  protected K generateActualKey(K key, V value) {
    return key;
  }

 class RuozedataMultipleTextOutputFormat extends MultipleTextOutputFormat[Any,Any]{
    override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = {
      s"$key/$name"
    }


    override def generateActualKey(key: Any, value: Any): AnyRef = {
      NullWritable.get()
    }

  }

你mapreduce代码里 不输出值 用什么？使用NullWritable.get()   不可能使用 “” 或者 null 你可以测试一下使用它们输出是什么。

object MulitOutputApp {

  def main(args: Array[String]): Unit = {
    val sc = ContextUtils.getSparkContext(this.getClass.getSimpleName)

    val output = "file:///C:/IdeaProjects/spark/out/mulit"
    val input = sc.textFile("file:///C:/IdeaProjects/spark/data/access.log")
    input.map(x => {
      val splits = x.split("\t")
      (splits(1), x) // (platform , 完整的日志)
    }).partitionBy(new HashPartitioner(5))
      .saveAsHadoopFile(output,classOf[String],classOf[String],classOf[RuozedataMultipleTextOutputFormat])
    sc.stop()
  }

  class RuozedataMultipleTextOutputFormat extends MultipleTextOutputFormat[Any,Any]{
    override def generateFileNameForKeyValue(key: Any, value: Any, name: String): String = {
      s"$key/$name"
    }

    override def generateActualKey(key: Any, value: Any): AnyRef = {
      NullWritable.get()
    }
  }
}

查看结果

在这里插入图片描述
基本功能实现完成