安科网

Spark入门（六）--Spark的combineByKey、sortBykey

登峰小蚁

登峰小蚁

2020-01-12

关注 关注

spark的combineByKey

combineByKey的特点

combineByKey的强大之处，在于提供了三个函数操作来操作一个函数。第一个函数，是对元数据处理，从而获得一个键值对。第二个函数，是对键值键值对进行一对一的操作，即一个键值对对应一个输出，且这里是根据key进行整合。第三个函数是对key相同的键值对进行操作，有点像reduceByKey，但真正实现又有着很大的不同。

在Spark入门（五）--Spark的reduce和reduceByKey中，我们用reduce进行求平均值。用combineByKey我们则可以求比平均值更为丰富的事情。现在有一个数据集，每一行数据包括一个a-z字母和一个整数，其中字母和整数之间以空格分隔。现在要求得每个字母的平均数。这个场景有点像多个学生，每个学生多门成绩，求得学生的平均分。但这里将问题简化，其中数据集放在grades中。数据集以及下面的代码都可以在github上下载。

combineByKey求多个平均值

scala实现

import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.{SparkConf, SparkContext}

object SparkCombineByKey {

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

    val conf = new SparkConf().setMaster("local").setAppName("SparkCombineByKey")

    val sc = new SparkContext(conf)

    sc.textFile("./grades").map(line=>{
      val splits = line.split(" ")
      (splits(0),splits(1).toInt)
    }).combineByKey(
      value => (value,1),
      (x:(Int,Int),y)=>(x._1+y,x._2+1),
      (x:(Int,Int),y:(Int,Int))=>(x._1+y._1,x._2+y._2)
    ).map(x=>(x._1,x._2._1/x._2._2)).foreach(println)

  }

}

scala运行结果

(d,338451)
(e,335306)
(a,336184)
(i,346279)
(b,333069)
(h,334343)
(f,341380)
(j,320145)
(g,334042)
(c,325022)
Spark入门（六）--Spark的combineByKey、sortBykey

java实现：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.sources.In;
import scala.Tuple2;

public class SparkCombineByKeyJava {

    public static void main(String[] args){

        SparkConf conf = new SparkConf().setMaster("local").setAppName("SparkCombineByKeyJava");

        JavaSparkContext sc = new JavaSparkContext(conf);

        combineByKeyJava(sc);

        combineByKeyJava8(sc);


    }


    public static void combineByKeyJava(JavaSparkContext sc){

        JavaPairRDD<String,Integer> splitData = sc.textFile("./grades").mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                String[] splits = s.split(" ");
                return new Tuple2<>(splits[0],Integer.parseInt(splits[1]));
            }
        });

        splitData.combineByKey(new Function<Integer, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> call(Integer integer) throws Exception {
                return new Tuple2<>(integer, 1);
            }
        }, new Function2<Tuple2<Integer, Integer>, Integer, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> integerIntegerTuple2, Integer integer) throws Exception {
                return new Tuple2<>(integerIntegerTuple2._1 + integer, integerIntegerTuple2._2 + 1);
            }
        }, new Function2<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> integerIntegerTuple2, Tuple2<Integer, Integer> integerIntegerTuple22) throws Exception {
                return new Tuple2<>(integerIntegerTuple2._1+integerIntegerTuple22._1,integerIntegerTuple2._2+integerIntegerTuple22._2);
            }
        }).map(new Function<Tuple2<String,Tuple2<Integer,Integer>>, Tuple2<String,Double>>() {
            @Override
            public Tuple2<String,Double> call(Tuple2<String, Tuple2<Integer, Integer>> stringTuple2Tuple2) throws Exception {
                return new Tuple2<>(stringTuple2Tuple2._1,stringTuple2Tuple2._2._1*1.0/stringTuple2Tuple2._2._2);
            }
        }).foreach(new VoidFunction<Tuple2<String, Double>>() {
            @Override
            public void call(Tuple2<String, Double> stringDoubleTuple2) throws Exception {
                System.out.println(stringDoubleTuple2._1+" "+stringDoubleTuple2._2);
            }
        });

    }

    public static void combineByKeyJava8(JavaSparkContext sc){

        JavaPairRDD<String,Integer> splitData = sc.textFile("./grades").mapToPair(line -> {
            String[] splits = line.split(" ");
            return new Tuple2<>(splits[0],Integer.parseInt(splits[1]));
        });

        splitData.combineByKey(
                x->new Tuple2<>(x,1),
                (x,y)->new Tuple2<>(x._1+y,x._2+1),
                (x,y)->new Tuple2<>(x._1+y._1,x._2+y._2)
        ).map(x->new Tuple2(x._1,x._2._1*1.0/x._2._2)).foreach(x->System.out.println(x._1+" "+x._2));

    }
}

java运行结果

d 338451.6
e 335306.7480769231
a 336184.95321637427
i 346279.497029703
b 333069.8589473684
h 334343.75
f 341380.94444444444
j 320145.7618069815
g 334042.37605042016
c 325022.4183673469
Spark入门（六）--Spark的combineByKey、sortBykey

分析

在开始python之前，我们先观察java和scala两个程序。我们发现java7的代码非常冗余，而java8和scala则相比起来非常干净利落。当然，我们难说好坏，但是这也表现出当代语言开始从繁就简的一个转变。到了python这一特点就体现的更加淋漓尽致。

但我们不光说语言，我们分析这个求平均的实现方式，由于java中对数值做了一个处理，因此有保留小数，而scala则没有，但至少可以判断两者的结果是一致的。当然，这不是重点，重点是，这个combinByKey非常复杂，有三个函数。我们很难观察到每个过程做了什么。因此我们在这里，对scala程序进行进一步的输出，从而观察combineByKey到底做了什么。

scala修改

import org.apache.spark.{SparkConf, SparkContext}

object SparkCombineByKey {

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

    val conf = new SparkConf().setMaster("local").setAppName("SparkCombineByKey")

    val sc = new SparkContext(conf)

    sc.textFile("./grades").map(line=>{
      val splits = line.split(" ")
      (splits(0),splits(1).toInt)
    }).combineByKey(
      value => {
        println("这是第一个函数")
        println("将所有的值遍历，并放在元组中，标记1")
        println(value)
        (value,1)
      },
      (x:(Int,Int),y)=>{
        println("这是第二个函数")
        println("将x中的第一个值进行累加求和，第二个值加一，求得元素总个数")
        println("x:"+x.toString())
        println("y:"+y)
        (x._1+y,x._2+1)
      },
      (x:(Int,Int),y:(Int,Int))=>{
        (x._1+y._1,x._2+y._2)
      }
    ).map(x=>(x._1,x._2._1/x._2._2)).foreach(println)

  }

}

得到结果

这是第一个函数
将所有的值遍历，并放在元组中，标记1
222783
这是第一个函数
将所有的值遍历，并放在元组中，标记1
48364
这是第一个函数
将所有的值遍历，并放在元组中，标记1
204950
这是第一个函数
将所有的值遍历，并放在元组中，标记1
261777
...
...
...
这是第二个函数
将x中的第一个值进行累加求和，第二个值加一，求得元素总个数
x:(554875,2)
y:357748
这是第二个函数
将x中的第一个值进行累加求和，第二个值加一，求得元素总个数
x:(912623,3)
y:202407
这是第一个函数
将所有的值遍历，并放在元组中，标记1
48608
这是第二个函数
将x中的第一个值进行累加求和，第二个值加一，求得元素总个数
x:(1115030,4)
y:69003
这是第一个函数
将所有的值遍历，并放在元组中，标记1
476893

...
...
...

(d,338451)
(e,335306)
(a,336184)
(i,346279)
(b,333069)
(h,334343)
(f,341380)
(j,320145)
(g,334042)
(c,325022)

这里我们发现了，函数的顺序并不先全部执行完第一个函数，再执行第二个函数。而是分区并行，即第一个分区执行完第一个函数，并不等待其他分区执行完第一个函数，而是紧接着执行第二个函数，最后在第三个函数进行处理。在本地单机下，该并行特点并不能充分发挥，但在集群环境中，各个分区在不同节点计算，然后处理完结果汇总处理。这样，当数据量十分庞大时，集群节点数越多，该优势就表现地越明显。

此外还有一个非常值得关注的特点，当我们把foreach(println)这句话去掉时

foreach(println)

我们运行程序，发现程序没有任何输出。这是由于spark的懒加载特点，spark只用在对数据执行具体操作时，如输出、保存等才会执行计算。这看起来有点不合理，但实际上这样做在很多场景下能大幅度提升效率，但如果没有处理好，可能会导致spark每次执行操作都会从头开始计算该过程。因此当一个操作结果需要被频繁或者多次调用的时候，我们应该将结果存下来。

python实现

from pyspark import SparkConf,SparkContext

conf = SparkConf().setMaster("local").setAppName("SparkCombineByKey")

sc = SparkContext(conf=conf)

sc.textFile("./grades")    .map(lambda line : (line.split(" ")[0],int(line.split(" ")[1])))    .combineByKey(
    lambda num:(num,1),lambda x,y:(x[0]+y,x[1]+1),lambda x,y:(x[0]+y[0],x[1]+y[1])
).map(lambda x:(x[0],x[1][0]/x[1][1])).foreach(print)

得到结果

(‘b‘, 333069.8589473684)
(‘f‘, 341380.94444444444)
(‘j‘, 320145.7618069815)
(‘h‘, 334343.75)
(‘a‘, 336184.95321637427)
(‘g‘, 334042.37605042016)
(‘d‘, 338451.6)
(‘e‘, 335306.7480769231)
(‘c‘, 325022.4183673469)
Spark入门（六）--Spark的combineByKey、sortBykey

spark的sortByKey

sortByKey进行排序

sortByKey非常简单，也非常常用。这里依然采用上述文本，将处理后的结果，进行排序，得到平均值最大的字母。在实际运用中我们这里可以看成求得按照成绩排序，或者按照姓名排序。

scala实现

import org.apache.spark.{SparkConf, SparkContext}

object SparkSortByKey {

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

    val conf = new SparkConf().setMaster("local").setAppName("SparkCombineByKey")

    val sc = new SparkContext(conf)


    val result = sc.textFile("./grades").map(line=>{
      val splits = line.split(" ")
      (splits(0),splits(1).toInt)
    }).combineByKey(value =>(value,1),(x:(Int,Int),y)=>(x._1+y,x._2+1),(x:(Int,Int),y:(Int,Int))=>(x._1+y._1,x._2+y._2)
    ).map(x=>(x._1,x._2._1/x._2._2))

    //按照名字排序，顺序
    result.sortByKey(true).foreach(println)

    //按照名字排序，倒序
    result.sortByKey(false).foreach(println)


    val result1 = sc.textFile("./grades").map(line=>{
      val splits = line.split(" ")
      (splits(0),splits(1).toInt)
    }).combineByKey(value =>(value,1),(x:(Int,Int),y)=>(x._1+y,x._2+1),(x:(Int,Int),y:(Int,Int))=>(x._1+y._1,x._2+y._2)
    ).map(x=>(x._2._1/x._2._2,x._1))

    //按照成绩排序，顺序
    result1.sortByKey(true).foreach(println)

    //按照成绩排序，倒序
    result1.sortByKey(false).foreach(println)


  }

}

python实现

from pyspark import SparkConf,SparkContext

conf = SparkConf().setMaster("local").setAppName("SparkCombineByKey")

sc = SparkContext(conf=conf)

result = sc.textFile("./grades")    .map(lambda line : (line.split(" ")[0],int(line.split(" ")[1])))    .combineByKey(
    lambda num:(num,1),lambda x,y:(x[0]+y,x[1]+1),lambda x,y:(x[0]+y[0],x[1]+y[1])
).map(lambda x:(x[0],x[1][0]/x[1][1]))

result.sortByKey(True).foreach(print)

result.sortByKey(False).foreach(print)


result1 = sc.textFile("./grades")    .map(lambda line : (line.split(" ")[0],int(line.split(" ")[1])))    .combineByKey(
    lambda num:(num,1),lambda x,y:(x[0]+y,x[1]+1),lambda x,y:(x[0]+y[0],x[1]+y[1])
).map(lambda x:(x[1][0]/x[1][1],x[0]))

result1.sortByKey(True).foreach(print)

result1.sortByKey(False).foreach(print)

得到结果

(a,336184)
(b,333069)
(c,325022)
(d,338451)
(e,335306)
(f,341380)
(g,334042)
(h,334343)
(i,346279)
(j,320145)



(j,320145)
(i,346279)
(h,334343)
(g,334042)
(f,341380)
(e,335306)
(d,338451)
(c,325022)
(b,333069)
(a,336184)


(320145,j)
(325022,c)
(333069,b)
(334042,g)
(334343,h)
(335306,e)
(336184,a)
(338451,d)
(341380,f)
(346279,i)


(346279,i)
(341380,f)
(338451,d)
(336184,a)
(335306,e)
(334343,h)
(334042,g)
(333069,b)
(325022,c)
(320145,j)

数据集以及代码都可以在github上下载。



转自：https://juejin.im/post/5c7a58426fb9a049a42fc190

spark 键值 apache 数据处理

登峰小蚁

登峰小蚁

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

Apache Spark有哪些局限性

Apache Spark是行业中流行和广泛使用的大数据工具之一。Apache Spark已成为业界的热门话题，并且如今非常流行。但工业正在转移朝向apache flink。Apache Spark是为快速计算而设计的开源，闪电般快速的集群计算框架。Apac

sxyhetao 2020-06-12

使用经 EMRFS S3 优化的提交器提高 Apache Spark 写入 Apache Parquet 格式文件的性能

使用经 EMRFS S3 优化的提交器提高 Apache Spark 写入 Apache Parquet 格式文件的性能。经 EMRFS S3 优化的提交程序是一款新的输出提交程序，可用于Amazon EMR5.19.0 及更高版本的Apache Spar

hovermenu 2020-06-10

spark系列之基本概念

　　RDD是对象的分布式集合。　　RDD也提供数据沿袭——以图形形式给出每个中间步骤的祖先树，当RDD的一个分区丢失，可以根据祖先树重建该分区。　　RDD有两组操作，转换和行动，RDD转换是有惰性的，宽窄依赖。　　SparkSession是读取数据、处理元

Johnson0 2020-07-28

Spark RDD

②通过Spark读取外部存储文件，将文件数据转变为RDD。可以从本地磁盘读取，也可以从HDFS读取

adayan0 2020-07-05

Spark Streaming读取Kafka数据两种方式

Receiver从Kafka中获取的数据都是存储在Spark Executor的内存中的，然后Spark Streaming启动的job会去处理那些数据。在提交Spark Streaming任务后，Spark集群会划出指定的Receivers来专门、持续不

rongwenbin 2020-06-15

Mr与spark的shuffle过程详解及对比

大数据的分布式计算框架目前使用的最多的就是hadoop的mapReduce和Spark，mapReducehe和Spark之间的最大区别是前者较偏向于离线处理，而后者重视实现性，下面主要介绍mapReducehe和Spark两者的shuffle过程。Map

Johnson0 2020-06-08

Spark分区

在Spark中，RDD是其最基本的抽象数据集，其中每个RDD是由若干个Partition组成。在Job运行期间，参与运算的Partition数据分布在多台机器的内存当中。图一中，RDD1包含了5个Partition，RDD2包含了3个Partition，这

Hhanwen 2020-05-29

Spark Streaming企业运用

========== Spark Streaming 是什么 ==========1、SPark Streaming 是 Spark 中一个组件，基于 Spark Core 进行构建，用于对流式进行处理，类似于 Storm。kafka、flume、HDFS

Hhanwen 2020-05-29

spark 任务提交参数

网上看到的关于Executor，Cores和Memory的分配相关博客，先记录下来，再汇总。Spark处理1Tb数据不需要1Tb的内存。具体需要多少内存是根据executor的core数量和每次读取数据集的block大小决定的。以读取hdfs上1tb大文件

adayan0 2020-05-19

Spark 数据分析调优 （一）

假设你已经清楚了Spark 的 RDD 转换，Action 等内容。并且已经知道用web UI来理解为什么Job要花这么长时间时，Job、stage和task 也很清楚了。在本文中，您将了解Spark程序在集群中实际执行的基础知识。Spark应用程序由单个

Johnson0 2020-05-17

【译】Using .NET for Apache Spark to Analyze Log Data

.NET for Spark可用于处理成批数据、实时流、机器学习和ad-hoc查询。在这篇博客文章中，我们将探讨如何使用.NET for Spark执行一个非常流行的大数据任务，即日志分析。日志分析的目标是从这些日志中获得有关工具或服务的活动和性能的有意义

登峰小蚁 2020-05-11

spark系列-5、RDD、DataFrame、Dataset的区别和各自的优势

从一开始 RDD 就是 Spark 提供的面向用户的主要 API。从根本上来说，一个 RDD 就是你的数据的一个不可变的分布式元素集合，在集群中跨节点分布，可以通过若干提供了转换和处理的底层 API 进行并行处理。但与 RDD 不同的是，数据都被组织到有名

Hhanwen 2020-05-04

spark工作原理

它是被分区的，分为多个分区，每个分区分布在集群中的不同节点上，从而让RDD中的数据可以被并行操作。即如果某个节点上的RDD partition，因为节点故障，导致数据丢了，那么RDD会自动通过自己的数据来源重新计算该partition。

Hhanwen 2020-05-03

spark streaming 读取kafka两种方式的区别

1】Receiver是使用高层次的consumer Api来实现的。receiver 接收的消息都是存储在spark Executor中的，然后spark启动jobq去处理那些消息。然而，默认情况下，这种方式会因为底层的失败丢失数据。如果要启用高可靠机制，

Oeljeklaus 2020-04-20

Spark SQL(6) OptimizedPlan

在这一步spark sql主要应用一些规则，优化生成的Resolved Plan，这一步涉及到的有Optimizer。之前介绍在sparksession实例化的是会实例化sessionState，进而确定QueryExecution、Analyzer，Op

Hhanwen 2020-07-26

spark 广播变量 累加器

你还可以使用 destroy 方法彻底销毁广播变量，调用该方法后，如果计算任务中又用到广播变量，则会抛出异常。　　广播变量在一定数据量范围内可以有效地使作业避免 Shuffle，使计算尽可能本地运行，Spark 的 Map 端连接操作就是用广播变量实现的。

zhixingheyitian 2020-07-19

Spark 源码解读（五）SparkContext的初始化之创建和启动DAGScheduler

DAGScheduler主要用于在任务正式提交给TaskSchedulerImpl提交之前做一些准备工作，包括：创建job,将DAG中的RDD划分到不同的Stage,提交Stage等等。SparkContext中创建DAGScheduler的代码如下所示：

yanqianglifei 2020-07-07

Spark DAG 依赖关系 Stage

记录了RDD之间的依赖关系，即RDD是通过何种变换生成的，如下图：RDD1是RDD2的父RDD，通过flatMap操作生成 借助RDD之间的依赖关系，可以实现数据的容错，但是已经尽力避免产生shuffle. Task任务对应的是分区，即一个分区就是一个Ta

Hhanwen 2020-07-05

Notebook Docker 安装spark环境

zhixingheyitian 2020-07-04

入门大数据---Spark开发环境搭建

Local 模式是最简单的一种运行方式，它采用单节点多线程方式运行，不用部署，开箱即用，适合日常测试开发。进入 spark-shell 后，程序已经自动创建好了上下文 SparkContext，等效于执行了下面的 Scala 代码：。安装完成后可以先做一个

Hhanwen 2020-06-25
登峰小蚁

登峰小蚁

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号