绑定完请刷新页面
取消
刷新
登录 | 注册

分享好友

×
取消 复制
SparkCore入门实战 （二）
2020-05-09 16:21:09

一、键值对RDD数据分区器

键值对RDD数据分区器

Spark目前支持Hash分区和Range分区，用户也可以自定义分区，Hash分区为当前的默认分区，Spark中分区器直接决定了RDD中分区的个数、RDD中每条数据经过Shuffle过程属于哪个分区和Reduce的个数

注意：

(1)只有Key-Value类型的RDD才有分区器的，非Key-Value类型的RDD分区器的值是None
(2)每个RDD的分区ID范围：0~numPartitions-1，决定这个值是属于那个分区的。

1、获取RDD分区

可以通过使用RDD的partitioner 属性来获取 RDD 的分区方式。它会返回一个 scala.Option 对象， 通过get方法获取其中的值。相关源码如下：

def getPartition(key: Any): Int = key match {
  case null => 
  case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
}
def nonNegativeMod(x: Int, mod: Int): Int = {
  val rawMod = x % mod
  rawMod + (if (rawMod < ) mod else )
}

（1）创建一个pairRDD

scala> val pairs = sc.parallelize(List((1,1),(2,2),(3,3)))
pairs: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[14] at parallelize at <console>:24

（2）查看RDD的分区器

scala> pairs.partitioner
res13: Option[org.apache.spark.Partitioner] = None

（3）导入HashPartitioner类

scala>  import org.apache.spark.HashPartitioner
import org.apache.spark.HashPartitioner

（4）使用HashPartitioner对RDD进行重新分区

scala> val partitioned = pairs.partitionBy(new HashPartitioner(2))
partitioned: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[15] at partitionBy at <console>:27

（5）查看重新分区后RDD的分区器

scala> partitioned.partitioner
res14: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@2)

2、Hash分区

HashPartitioner分区的原理：对于给定的key，计算其hashCode，并除以分区的个数取余，如果余数小于0，则用余数+分区的个数（否则加0），后返回的值就是这个key所属的分区ID。

使用Hash分区的实操

scala> val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)
nopar: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[16] at parallelize at <console>:25

scala> nopar.mapPartitionsWithIndex((index,iter)=>{ Iterator(index.toString+" : "+iter.mkString("|")) }).collect
res16: Array[String] = Array("0 : ", 1 : (1,3), 2 : (1,2), 3 : (2,4), "4 : ", 5 : (2,3), 6 : (3,6), 7 : (3,8))

scala>  val hashpar = nopar.partitionBy(new org.apache.spark.HashPartitioner(7))
hashpar: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[18] at partitionBy at <console>:27

scala> hashpar.count
res17: Long = 6                                                                 

scala>  hashpar.partitioner
res18: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@7)

scala> hashpar.mapPartitions(iter => Iterator(iter.length)).collect()
res19: Array[Int] = Array(, 2, 2, 2, , , )

scala> nopar.partitioner
res20: Option[org.apache.spark.Partitioner] = None


3、 Ranger分区

HashPartitioner分区弊端：可能导致每个分区中数据量的不均匀，极端情况下会导致某些分区拥有RDD的全部数据。
RangePartitioner作用：将一定范围内的数映射到某一个分区内，尽量保证每个分区中数据量的均匀，而且分区与分区之间是有序的，一个分区中的元素肯定都是比另一个分区内的元素小或者大，但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。实现过程为：

步：先重整个RDD中抽取出样本数据，将样本数据排序，计算出每个分区的大key值，形成一个Array[KEY]类型的数组变量rangeBounds；

第二步：判断key在rangeBounds中所处的范围，给出该key值在下一个RDD中的分区id下标；该分区器要求RDD中的KEY类型必须是可以排序的


4、自定义分区

要实现自定义的分区器，你需要继承 org.apache.spark.Partitioner 类并实现下面三个方法。

（1）numPartitions: Int:返回创建出来的分区数。
（2）getPartition(key: Any): Int:返回给定键的分区编号(0到numPartitions-1)。
（3）equals():Java 判断相等性的标准方法。这个方法的实现非常重要，Spark 需要用这个方法来检查你的分区器对象是否和其他分区器实例相同，这样 Spark 才可以判断两个 RDD 的分区方式是否相同。

需求：将相同后缀的数据写入相同的文件，通过将相同后缀的数据分区到相同的分区并保存输出来实现。

（1）创建一个pairRDD

scala> val data = sc.parallelize(Array((1,1),(2,2),(3,3),(4,4),(5,5),(6,6)))
data: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[20] at parallelize at <console>:25

（2）定义一个自定义分区类

scala> :paste
// Entering paste mode (ctrl-D to finish)
class CustomerPartitioner(numParts:Int) extends org.apache.spark.Partitioner{

  //覆盖分区数
  override def numPartitions: Int = numParts

  //覆盖分区号获取函数
  override def getPartition(key: Any): Int = {
    val ckey: String = key.toString
    ckey.substring(ckey.length-1).toInt%numParts
  }
}

// Exiting paste mode, now interpreting.

defined class CustomerPartitioner

（3）将RDD使用自定义的分区类进行重新分区

scala> val par = data.partitionBy(new CustomerPartitioner(2))
par: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[2] at partitionBy at <console>:27

（4）查看重新分区后的数据分布

scala> par.mapPartitionsWithIndex((index,items)=>items.map((index,_))).collect
res3: Array[(Int, (Int, Int))] = Array((,(2,2)), (,(4,4)), (,(6,6)), (1,(1,1)), (1,(3,3)), (1,(5,5)))

使用自定义的 Partitioner 是很容易的:只要把它传给 partitionBy() 方法即可。Spark 中有许多依赖于数据混洗的方法，比如 join() 和 groupByKey()，它们也可以接收一个可选的 Partitioner 对象来控制输出数据的分区方式。


二、Spark连接HBase数据读取与保存

数据读取与保存

Spark的数据读取及数据保存可以从两个维度来作区分：文件格式以及文件系统。
文件格式分为：Text文件、Json文件、Csv文件、Sequence文件以及Object文件；
文件系统分为：本地文件系统、HDFS、HBASE以及数据库

1、文件类数据读取与保存

Text文件

（1）数据读取:textFile(String)

scala> val hdfsFile = sc.textFile("hdfs://hadoop105:9000/fruit.txt")

hdfsFile: org.apache.spark.rdd.RDD[String] = hdfs://hadoop102:9000/fruit.txt MapPartitionsRDD[21] at textFile at <console>:24

（2）数据保存: saveAsTextFile(String)

scala> hdfsFile.saveAsTextFile("/fruitOut")

Json文件

如果JSON文件中每一行就是一个JSON记录，那么可以通过将JSON文件当做文本文件来读取，然后利用相关的JSON库对每一条数据进行JSON解析。
注意：使用RDD读取JSON文件处理很复杂，同时SparkSQL集成了很好的处理JSON文件的方式，所以应用中多是采用SparkSQL处理JSON文件。

准备文件数据：

（1）在in文件夹下创建user.json文件数据，编辑内容：

{"name":"123","age": 20}
{"name":"456","age": 20}
{"name":"789","age": 20}

方式一

代码实现：

package com.study.bigdatabase

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.mortbay.util.ajax.JSON
//检查点
object Spark06_RDD_Serializable {
  def main(args: Array[String]): Unit = {

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")

    //创建Spark上下文对象
    val sc = new SparkContext(config)
   //读取文件
    val json = sc.textFile("in/user.json")

    //解析json数据
    val result  = json.map(JSON.parse)
    result.foreach(println)
    //释放资源
    sc.stop()
  }
}

启动程序，控制台打印：

方式二：

命令方式：

（1）导入解析json所需的包

scala> import scala.util.parsing.json.JSON

（2）上传json文件到HDFS

[root@hadoop105 spark]$ hadoop fs -put ./examples/src/main/resources/people.json /

（3）读取文件

scala> val json = sc.textFile("/people.json")
json: org.apache.spark.rdd.RDD[String] = /people.json MapPartitionsRDD[8] at textFile at <console>:24

（4）解析json数据

scala> val result  = json.map(JSON.parseFull)
result: org.apache.spark.rdd.RDD[Option[Any]] = MapPartitionsRDD[10] at map at <console>:27

（5）打印

scala> result.collect
res11: Array[Option[Any]] = Array(Some(Map(name -> Michael)), Some(Map(name -> Andy, age -> 30.)), Some(Map(name -> Justin, age -> 19.)))



2、 Sequence文件

SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。Spark 有专门用来读取 SequenceFile 的接口。在 SparkContext 中，可以调用 sequenceFile keyClass, valueClass。
注意：SequenceFile文件只针对PairRDD

（1）创建一个RDD

scala> val rdd = sc.parallelize(Array((1,2),(3,4),(5,6)))
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[24] at parallelize at <console>:26

（2）将RDD保存为Sequence文件

scala> rdd.saveAsSequenceFile("file:///opt/module/spark/seqFile")

（3）查看该文件

[root@hadoop105 seqFile]$ pwd
/opt/module/spark/seqFile

[root@hadoop105 seqFile]$ ll
总用量 8
-rw-r--r-- 1 atguigu atguigu 108 10月  9 10:29 part-00000
-rw-r--r-- 1 atguigu atguigu 124 10月  9 10:29 part-00001
-rw-r--r-- 1 atguigu atguigu    10月  9 10:29 _SUCCESS

[root@hadoop105 seqFile]$ cat part-00000
SEQ org.apache.hadoop.io.IntWritable org.apache.hadoop.io.IntWritable

（4）读取Sequence文件

scala> val seq = sc.sequenceFile[Int,Int]("file:///opt/module/spark/seqFile")
seq: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[18] at sequenceFile at <console>:24

（5）打印读取后的Sequence文件

scala> seq.collect
res14: Array[(Int, Int)] = Array((1,2), (3,4), (5,6))



3、对象文件

对象文件是将对象序列化后保存的文件，采用Java的序列化机制。可以通过objectFile [k,v] (path) 函数接收一个路径，读取对象文件，返回对应的 RDD，也可以通过调用saveAsObjectFile() 实现对对象文件的输出。因为是序列化所以要指定类型。

（1）创建一个RDD

scala> val rdd = sc.parallelize(Array(1,2,3,4))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[25] at parallelize at <console>:26

（2）将RDD保存为Object文件

scala> rdd.saveAsObjectFile("file:///opt/module/spark/objectFile")

（3）查看该文件

[root@hadoop105 objectFile]$ pwd
/opt/module/spark/objectFile

[root@hadoop105 objectFile]$ ll
总用量 8
-rw-r--r-- 1 atguigu atguigu 142 10月  9 10:37 part-00000
-rw-r--r-- 1 atguigu atguigu 142 10月  9 10:37 part-00001
-rw-r--r-- 1 atguigu atguigu    10月  9 10:37 _SUCCESS

[root@hadoop105 objectFile]$ cat part-00000 
SEQ!org.apache.hadoop.io.NullWritable"org.apache.hadoop.io.BytesWritableW@`l

（4）读取Object文件

scala> val objFile = sc.objectFile[Int]("file:///opt/module/spark/objectFile")

objFile: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[31] at objectFile at <console>:24

（5）打印读取后的Sequence文件

scala> objFile.collect
res19: Array[Int] = Array(1, 2, 3, 4)



4、文件系统类数据读取与保存HDFS

Spark的整个生态系统与Hadoop是完全兼容的,所以对于Hadoop所支持的文件类型或者数据库类型,Spark也同样支持.另外,由于Hadoop的API有新旧两个版本,所以Spark为了能够兼容Hadoop所有的版本,也提供了两套创建操作接口.对于外部存储创建操作而言,hadoopRDD和newHadoopRDD是为抽象的两个函数接口,主要包含以下四个参数.

（1）输入格式(InputFormat):制定数据输入的类型,如TextInputFormat等,新旧两个版本所引用的版本分别是org.apache.hadoop.mapred.InputFormat和org.apache.hadoop.mapreduce.InputFormat(NewInputFormat)
（2）键类型: 指定[K,V]键值对中K的类型
（3）值类型: 指定[K,V]键值对中V的类型
（4）分区值:指定由外部存储生成的RDD的partition数量的小值,如果没有指定,系统会使用默认值defaultMinSplits注意:其他创建操作的API接口都是为了方便终的Spark程序开发者而设置的,是这两个接口的高效实现版本.例如,对于textFile而言,只有path这个指定文件路径的参数,其他参数在系统内部指定了默认值。

a、在Hadoop中以压缩形式存储的数据,不需要指定解压方式就能够进行读取,因为Hadoop本身有一个解压器会根据压缩文件的后缀推断解压算法进行解压.

b、如果用Spark从Hadoop中读取某种类型的数据不知道怎么读取的时候,上网查找一个使用map-reduce的时候是怎么读取这种这种数据的,然后再将对应的读取方式改写成上面的hadoopRDD和newAPIHadoopRDD两个类就行了



5、MySQL数据库连接

支持通过Java JDBC访问关系型数据库。需要通过JdbcRDD进行，示例如下:

数据库准备：

（1）添加依赖

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.27</version>
</dependency>

代码实现：

package com.study.bigdatabase

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

//检查点
object Spark07_RDD_MySql {
  def main(args: Array[String]): Unit = {

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")

    //创建Spark上下文对象
    val sc = new SparkContext(config)

    val driver="com.mysql.jdbc.Driver"
    var url="jdbc:mysql://hadoop105:3306/RDD"
    val userName="root"
    val password="123456"

    //创建 JDBCRDD，方法数据库
    var sql ="select name,age from user where id >= ? and id <= ?"
    val jdbcRDD = new JdbcRDD(
      sc,
      () => {
        //获取数据库连接对象
        Class.forName(driver)
        java.sql.DriverManager.getConnection(url, userName, password)
      },
      sql,
      1,
      3,
      2,
      
      (rs)=>{
        println(rs.getString(1)+","+rs.getInt(2))
      }
    )
    jdbcRDD.collect()

    //释放资源
    sc.stop()
  }
}

启动程序，控制台打印信息：

若代码中sql这样编写，会出错：

 var sql ="select name,age from user"

运行程序，查看信息：

保存数据

准备user表：空数据

代码实现：

package com.study.bigdatabase

import java.sql.Connection

import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

//检查点
object Spark07_RDD_MySql {
  def main(args: Array[String]): Unit = {

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")

    //创建Spark上下文对象
    val sc = new SparkContext(config)

   val driver="com.mysql.jdbc.Driver"
    var url="jdbc:mysql://hadoop105:3306/RDD"
    val userName="root"
    val password="123456"

    //保存数据
    val dataRDD:RDD[(String,Int)] = sc.makeRDD(List(("zhangsan", 20), ("lisi", 30), ("wangwu", 40)))

    dataRDD.foreach{
      case ( name,age ) =>{
        Class.forName(driver)
        val connection:Connection = java.sql.DriverManager.getConnection(url, userName, password)

        val sql="insert into user (name, age) values (?,?)"
        val statement = connection.prepareStatement(sql)
        statement.setString(1,name)
        statement.setInt(2,age)
        statement.executeUpdate()
        statement.close()
        connection.close()
      }
    }
    //释放资源
    sc.stop()
  }
}

启动程序运行，控制台打印信息：



6、HBase数据库

由于 org.apache.hadoop.hbase.mapreduce.TableInputFormat类的实现，Spark 可以通过Hadoop输入格式访问HBase。这个输入格式会返回键值对数据，其中键的类型为org. apache.hadoop.hbase.io.ImmutableBytesWritable，而值的类型为org.apache.hadoop.hbase.client.Result。

（1）添加依赖

<dependency>
	<groupId>org.apache.hbase</groupId>
	<artifactId>hbase-server</artifactId>
	<version>1.3.1</version>
</dependency>

<dependency>
	<groupId>org.apache.hbase</groupId>
	<artifactId>hbase-client</artifactId>
	<version>1.3.1</version>
</dependency>

(2) 分析源码图与步骤

拷贝在项目工程resource文件下

在编写代码之前，我们在Linux环境的Hbase创建一张rddtable表：

hbase(main):001:> create 'rddtable','info'
 row(s) in 1.7850 seconds

=> Hbase::Table - rddtable

随后，在这张表中，插入数据：

hbase(main):001:> put 'rddtable','1001','info:name',"zhanfsan"
 row(s) in .5160 seconds

Spark与HBase连接，代码实现：

package com.study.bigdatabase


import org.apache.hadoop.hbase
import org.apache.hadoop.hbase.{Cell, CellUtil, HBaseConfiguration}
import org.apache.hadoop.hbase.client.Result
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

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

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //创建Spark上下文对象
    val sc = new SparkContext(config)

    val conf = HBaseConfiguration.create()

    conf.set(TableInputFormat.INPUT_TABLE, "rddtable")

    val hbaseRDD:RDD[(ImmutableBytesWritable,Result)] = sc.newAPIHadoopRDD(
      conf,
      classOf[TableInputFormat],
      classOf[ImmutableBytesWritable],
      classOf[Result]
    )
    hbaseRDD.foreach{
      case (rowkey,result) => {
           //取数据
           val cells:Array[Cell] = result.rawCells()
        for (cell <- cells){
          println(Bytes.toString((CellUtil.cloneValue(cell))))
        }
    }
    }
    //释放资源
    sc.stop()
  }
}

启动程序，运行查看一下：



7、在HBase添加数据（插入数据）

代码实现：

package com.study.bigdatabase


import org.apache.hadoop.hbase.{Cell, CellUtil, HBaseConfiguration}
import org.apache.hadoop.hbase.client.{Put, Result}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

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

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //创建Spark上下文对象
    val sc = new SparkContext(config)

    val conf = HBaseConfiguration.create()

    val dataRDD:RDD[(String,String)] = sc.makeRDD(List(("1002", "zhangsan"), ("1003", "lisi"),
      ("1004", "wangwu")))

    val putRDD:RDD[(ImmutableBytesWritable,Put)] = dataRDD.map{
      case (rowkey,name) =>{


        val put = new Put(Bytes.toBytes(rowkey))
        put.addColumn(Bytes.toBytes("info"),Bytes.toBytes("name"),Bytes.toBytes(name))

        (new ImmutableBytesWritable(Bytes.toBytes(rowkey)),put)
      }
    }

    val jobConf = new JobConf(conf)
   jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE,"rddtable")

    putRDD.saveAsHadoopDataset(jobConf)

    //释放资源
    sc.stop()
  }
}

启动程序，运行查看一下：无报错

在HBase查看，扫描rddtable表，有数据了：

hbase(main):003:> scan 'rddtable'
ROW                                         COLUMN+CELL                                                                                                                    
 1001                                       column=info:name, timestamp=1581087691942, value=zhanfsan                                                                      
 1002                                       column=info:name, timestamp=1581112909827, value=zhangsan                                                                      
 1003                                       column=info:name, timestamp=1581112909815, value=lisi                                                                          
 1004                                       column=info:name, timestamp=1581112909792, value=wangwu                                                                        
4 row(s) in 4.5400 seconds


三、Spark 三大数据结构

RDD：分布式数据集
广播变量：分布式只读共享变量
累加器： 分布式只写共享变量

1、累加器

累加器用来对信息进行聚合，通常在向 Spark传递函数时，比如使用 map() 函数或者用 filter() 传条件时，可以使用驱动器程序中定义的变量，但是集群中运行的每个任务都会得到这些变量的一份新的副本，更新这些副本的值也不会影响驱动器中的对应变量。如果我们想实现所有分片处理时更新共享变量的功能，那么累加器可以实现我们想要的效果。

代码具体实现：

package com.study.bigdatabase

import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

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

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //创建Spark上下文对象
    val sc = new SparkContext(config)

    val dataRDD: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)

    /*种普通写法：
    val i: Int = dataRDD.reduce(_ + _)
     println(i)*/

    //第二种方法：使用累加器来共享变量
    var sum:Int =
    //（1）创建累加器对象
    val accumulator: LongAccumulator = sc.longAccumulator

    dataRDD.foreach{
      case i =>{
        //（2）执行累加器的累加功能
        accumulator.add(i)
      }
    }
    println("sum ="+accumulator.value)

    //释放资源
    sc.stop()
  }
}

启动程序运行，控制台打印信息：


2、自定义累加器

自定义累加器类型的功能在1.X版本中就已经提供了，但是使用起来比较麻烦，在2.0版本后，累加器的易用性有了较大的改进，而且官方还提供了一个新的抽象类：AccumulatorV2来提供更加友好的自定义类型累加器的实现方式。实现自定义类型累加器需要继承AccumulatorV2并至少覆写下例中出现的方法，下面这个累加器可以用于在程序运行过程中收集一些文本类信息，终以Set[String]的形式返回。

（1）分析源码图：

代码实现：

package com.study.bigdatabase

import java.util

import org.apache.spark.rdd.RDD
import org.apache.spark.util.{AccumulatorV2, LongAccumulator}
import org.apache.spark.{SparkConf, SparkContext}

//自定义累加器
object Spark10_ShareData {
  def main(args: Array[String]): Unit = {

    val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //创建Spark上下文对象
    val sc = new SparkContext(config)

    val dataRDD: RDD[String] = sc.makeRDD(List("hadoop", "hive", "hbase", "Scala","Spark"), 2)

  
    //  TODO  创建累加器
    val wordAccumnlator = new WordAccumnlator
    //  TODO  注册累加器
    sc.register(wordAccumnlator)

    dataRDD.foreach{
      case word =>{
        //  TODO  执行累加器累加功能
        wordAccumnlator.add(word)
      }
    }
    //  TODO  获取累加器的值
    println("sum ="+wordAccumnlator.value)

    //释放资源
    sc.stop()

  }
}

//声明累加器
//1、继承AccumulatorV2
//2、实现抽象方法
//3、创建累加器
class WordAccumnlator extends AccumulatorV2[String, util.ArrayList[String]]{

  val list = new util.ArrayList[String]()

  //当前累加器是否初始化状态
  override def isZero: Boolean = {
    list.isEmpty
  }

  //复制累加器对象
  override def copy(): AccumulatorV2[String, util.ArrayList[String]] = {
    new WordAccumnlator()
  }

  //重置累加器对象
  override def reset(): Unit = {
     list.clear()
  }

  //向累加器中增加数据
  override def add(v: String): Unit = {
    if (v.contains("h")) {
      list.add(v)
    }
  }

  //合并累加器
  override def merge(other: AccumulatorV2[String, util.ArrayList[String]]): Unit = {
    list.addAll(other.value)
  }

   //获取累加器的结果
  override def value: util.ArrayList[String] = list
}

启动程序，控制台打印信息：

以”h“累加聚中在一起

3、广播变量（调优策略）

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个Spark操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表，甚至是机器学习算法中的一个很大的特征向量，广播变量用起来都很顺手。 在多个并行操作中使用同一个变量，但是 Spark会为每个任务分别发送。
scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))
broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast()

scala>  broadcastVar.value
res0: Array[Int] = Array(1, 2, 3)

使用广播变量的过程如下：

(1) 通过对一个类型 T 的对象调用 SparkContext.broadcast 创建出一个 Broadcast[T] 对象。 任何可序列化的类型都可以这么实现。
(2) 通过 value 属性访问该对象的值(在 Java 中为 value() 方法)。
(3) 变量只会被发到各个节点一次，应作为只读值处理(修改这个值不会影响到别的节点)。


4、扩展

RDD相关概念关系

输入可能以多个文件的形式存储在HDFS上，每个File都包含了很多块，称为Block。当Spark读取这些文件作为输入时，会根据具体数据格式对应的InputFormat进行解析，一般是将若干个Block合并成一个输入分片，称为InputSplit，注意InputSplit不能跨越文件。随后将为这些输入分片生成具体的Task。InputSplit与Task是一一对应的关系。随后这些具体的Task每个都会被分配到集群上的某个节点的某个Executor去执行。

(1)每个节点可以起一个或多个Executor。

(2)每个Executor由若干core组成，每个Executor的每个core一次只能执行一个Task。

(3)每个Task执行的结果就是生成了目标RDD的一个partiton。

注意: 这里的core是虚拟的core而不是机器的物理CPU核，可以理解为就是Executor的一个工作线程。而 Task被执行的并发度 = Executor数目 * 每个Executor核数。至于partition的数目：

(1)对于数据读入阶段，例如sc.textFile，输入文件被划分为多少InputSplit就会需要多少初始Task。
(2)在Map阶段partition数目保持不变。
(3)在Reduce阶段，RDD的聚合会触发shuffle操作，聚合后的RDD的partition数目跟具体操作有关，例如repartition操作会聚合成指定分区数，还有一些算子是可配置的。

RDD在计算的时候，每个分区都会起一个task，所以rdd的分区数目决定了总的的task数目。申请的计算节点（Executor）数目和每个计算节点核数，决定了你同一时刻可以并行执行的task。

比如的RDD有100个分区，那么计算的时候就会生成100个task，你的资源配置为10个计算节点，每个两2个核，同一时刻可以并行的task数目为20，计算这个RDD就需要5个轮次。如果计算资源不变，你有101个task的话，就需要6个轮次，在后一轮中，只有一个task在执行，其余核都在空转。如果资源不变，你的RDD只有2个分区，那么同一时刻只有2个task运行，其余18个核空转，造成资源浪费。这就是在spark调优中，增大RDD分区数目，增大任务并行度的做法。

上一篇：2019，云原生技术的商业化元年
下一篇：搭建hbase集群环境
分享好友

分享这个小栈给你的朋友们，一起进步吧。

Apache Spark技术专区
创建时间：2020-05-08 17:16:40
Apache Spark是专为大规模数据处理而设计的快速通用的计算引擎 。现在形成一个高速发展应用广泛的生态系统。
展开
订阅须知

• 所有用户可根据关注领域订阅专区或所有专区

• 付费订阅：虚拟交易，一经交易不退款；若特殊情况，可3日内客服咨询

• 专区发布评论属默认订阅所评论专区（除付费小栈外）

技术专家

查看更多
  • 栈栈
    专家
戳我，来吐槽～

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8  北京市公安局海淀分局网监中心备案编号：11010802021510  未成年人举报专区

违法不良信息举报电话:18101376593   广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员