本文主要是介绍利用repartition和mapPartitions替代reduce功能，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
数据：用户，时间，地点 样例：
10001,20190401 14:20:06,20000000010000010000000000048793
10002,20190612 00:36:24,00000001000000050000000000181362
10002,20190612 01:49:05,00000001000000050000000000181362需求统计：用户在每个地点的停留时长 如果相邻记录的地点不一致，时长累加在前一个地点上。
正常流程是根据用户id分组，根据时间戳升序排列，遍历记录相邻记录时间戳相减，得到停留时长，再做累加。
这个需求正好符合spark里面 repartitionAndSortWithinPartitions 算子的使用功能。
import util.DateFormat
import org.apache.spark.{Partitioner, SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}object Test {def main(args: Array[String]): Unit = {//读取Spark Application 的配置信息val sparkConf = new SparkConf()//设置SparkApplication名称.setAppName("ModuleSpark Application").setMaster("local[2]")val sc = SparkContext.getOrCreate(sparkConf)val spark = SparkSession.builder.config(sparkConf).getOrCreate()val hadoopConf = sc.hadoopConfigurationval schema = StructType(Array(StructField("user_id", DataTypes.StringType),StructField("start_time", DataTypes.StringType),StructField("content_id", DataTypes.StringType)))//读入数据的时候可以直接加schema参数 同样可以利用csv直接切分val testRDD = spark.read.option("delimiter", ",").schema(schema).csv("localtest/input/userTimeLineDir/test.txt")//替换默认的排序 repartition中默认使用到orderingimplicit val my_self_Ordering = new Ordering[Tuple2[String,String]] {override def compare(a: Tuple2[String,String], b: Tuple2[String,String]): Int = {if(a._1 == b._1){a._2.compareTo(b._2)}else{a._1.compareTo(a._2)}}}//数据分区依据，分区时使用默认排序class KeyBasePartitioner(partitions: Int) extends Partitioner {//分区数override def numPartitions: Int = partitionsoverride def getPartition(key: Any): Int = {val k = key.asInstanceOf[Tuple2[String,String]]Math.abs(k._1.hashCode() % numPartitions)}}//特殊迭代器生成，直接生成结果数据的迭代器 尽量避免OOM//不可更改迭结果的记录条数class CustomIterator(iter: Iterator[((String,String),Row)]) extends Iterator[Row] {var lastUser:String = null;var lasttime:String = null;def hasNext : Boolean={iter.hasNext}def next :Row={val cur:((String,String),Row)=iter.nextval user = cur._2.getString(0)val contentid = cur._2.getString(2)val starttime = cur._2.getString(1)if(lastUser == null || lastUser != user ){lastUser = userlasttime = starttimeRow("3","4","5")}else{val dura = DateFormat.dateToTimestamp(starttime,"yyyyMMdd HH:mm:ss") - DateFormat.dateToTimestamp(lasttime,"yyyyMMdd HH:mm:ss")lastUser = userlasttime = starttimeRow(user,contentid,dura)}}}val result = testRDD.rdd.map( x =>((x.getString(0),x.getString(1)) ,x)).repartitionAndSortWithinPartitions(new KeyBasePartitioner(10)).mapPartitions(v => new CustomIterator(v))result.collect();result.saveAsTextFile("localtest/output")}
}1 利用spark的参数设置，直接生成固定格式的DataSet
spark.read.csv可以直接根据分隔符将读取的文件分列，通过设置option("delimiter",",")改变分隔符，通过.schema()直接设置读取的内容格式。
2 利用Ordering和Partitioner对数据进行重排序
排序工作在分区的时候一起进行，所有需要排序的内容都要并入key值。（疑惑点在这里，如果把时间戳并入key值，之后的reduce操作就需要重新map一次，重新分配key值userid，但在这个过程中，无法保证数据的排列顺序，而且正常reduce需要再一次触发shuffle操作，达不到优化的效果，所以这边选择采用MapPartitions来避免reduce。）
Ordering的排序是默认进行的，当重写了Ordering方法之后，默认使用重写后的方法进行排序。代码中按userid降序时间戳升序。
repartitionAndSortWithinPartitions对原始数据进行重新分区，分区时用到了自定义的Parititioner，注意分区时只用了userid字段，字段顺序按照定义的Ordering排列。
3 利用MapPartitions遍历记录，替换reduce效果
Map和MapPartitions区别：map算子中的执行命令针对每一条记录调用一次（不确定有没有内部优化），mappartitions算子针对每个分区记录调用一次。一般来说，使用MapPartitions的时候先获取一个包含分区内所有记录的迭代器，依次遍历，计算结果存储到List等集合容器，返回容器的迭代器。整个过程中如果分区中结果数据的数据量量太大，容易爆内存。但是MapPartitions的优点在于，可以很方便的使用外部变量，减少初始化的次数，也可以联系上下记录。
例如在计算停留时长，外部变量保留上条记录的用户编号和时间戳，如果用户编号相同，时间戳相减；用户编号不同，更新用户编号和时间戳（正常可以通过迭代器直接返回需要的结果，问题是分区内包含大量不同用户的数据，如果数据量过大，可能爆内存）。
有个替换方法是，继承迭代器的子类，传入原始数据（MapPartitions的迭代器），直接生成包含正确结果的迭代器返回，避免使用存储结果的集合容器。（不过这样做的缺点是，返回的数据没有办法控制记录条数，和输入的记录条数一模一样，如果是删除某些记录，可以用filter，但如果是增加部分记录，没想到处理办法。）
参考文章
你真知道如何高效用mapPartitions吗？ https://blog.csdn.net/rlnLo2pNEfx9c/article/details/81613035
How to Use Spark Transformations Efficiently for MapReduce-like Jobs https://technology.finra.org/code/using-spark-transformations-for-mpreduce-jobs.html
这篇关于利用repartition和mapPartitions替代reduce功能的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
