Spark基础

优化的执行计划：查询计划通过Spark catalyst optimiser进行优化

// RDD转换为DF，DS时，需要增加隐式转换，需要引入spark环境对象的隐式转换规则

import spark.implicits._

val df = rdd.toDF("id","name","age")

val rdd1 = df.rdd

val userRDD = rdd.map {

case (id, name, age) => {

User(id, name, age)

}

}

分支主题

val userDS = userRDD.toDS()

val rdd3 = userDS.rdd

val ds = df.as[User]

val df1 = ds.toDF()

三者均是Spark平台下的分布式弹性数据集

都有惰性机制，Action时才运算

根据内存，自动缓存运算

均有partition概念

import spark.implicits._，需要导入此包支持操作

RDD不支持SparkSQL

与RDD和Dataset不同，DataFrame每一行的类型固定为Row，每一列的值没法直接访问，只有通过解析才能获取各个字段的值

Dataset和DataFrame拥有完全相同的成员函数，区别只是每一行的数据类型不同

修改配置项spark.sql.sources.default，可修改默认数据源格式

当数据源格式不是parquet格式文件时，需要手动指定数据源的格式。数据源格式需要指定全名（例如：org.apache.spark.sql.parquet）

如果数据源格式为内置格式，则只需要指定简称定json, parquet, jdbc, orc, libsvm, csv, text来指定数据的格式

class CustomerReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {

分支主题

//最初启动的时候，调用该方法，作用为：读数据并将数据发送给Spark

override def onStart(): Unit = {

new Thread("Socket Receiver") {

override def run() {

receive()

}

}.start()

}

分支主题

//读数据并将数据发送给Spark

def receive(): Unit = {

分支主题

//创建一个Socket

var socket: Socket = new Socket(host, port)

分支主题

//定义一个变量，用来接收端口传过来的数据

var input: String = null

分支主题

//创建一个BufferedReader用于读取端口传来的数据

val reader = new BufferedReader(new InputStreamReader(socket.getInputStream, StandardCharsets.UTF_8))

分支主题

//读取数据

input = reader.readLine()

分支主题

//当receiver没有关闭并且输入数据不为空，则循环发送数据给Spark

while (!isStopped() && input != null) {

store(input)

input = reader.readLine()

}

分支主题

//跳出循环则关闭资源

reader.close()

socket.close()

分支主题

//重启任务

restart("restart")

}

分支主题

override def onStop(): Unit = {}

}

<dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-streaming-kafka-0-8_2.11</artifactId>

<version>2.1.1</version>

</dependency>

<dependency>

<groupId>org.apache.kafka</groupId>

<artifactId>kafka-clients</artifactId>

<version>0.11.0.2</version>

</dependency>

所有计算运行在一个线程中

指定使用几个线程来运行计算，通常CPU有几核，就指定几个线程

按CPU内部最多的Cores设置线程数

自动进行内存和磁盘数据存储的切换

基于Lineage（血统）的高效容错机制

Task如果失败，自动进行特定次数的重试

Stage如果失败，自动进行特定次数的重试

checkpoint和persist，可主动或被动触发

数据调度弹性：DAGScheduler、TaskScheduler与资源管理无关

数据分片的高度弹性（coalesce）

数据的来源

数据的类型 & 计算逻辑的封装 （数据模型）

计算逻辑不可变

提高数据处理能力

多任务同时执行

如果RDD是通过已有的文件系统构建，则compute函数是读取指定文件系统中的数据

如果RDD是通过其他RDD转换而来，则compute函数是执行转换逻辑将其他RDD的数据进行转换

RDD的操作算子包括两类，一类叫做transformations，它是用来将RDD进行转化，构建RDD的血缘关系

一类叫做actions，它是用来触发RDD的计算，得到RDD的相关计算结果或者将RDD保存的文件系统中

窄依赖

宽依赖

分支主题

返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成

类似于map，但独立地在RDD的每一个分片（分区）上运行.

map()：每次处理一条数据

mapPartitions()：每次处理一个分区的数据，这个分区的数据处理完后，原RDD中分区的数据才能释放，可能导致OOM

当内存空间较大的时候建议使用mapPartitions()，以提高处理效率

类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U]

类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）

将每一个分区形成一个数组，形成新的RDD类型时RDD[Array[T]]

分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器

过滤。返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成。

以指定的随机种子随机抽样出数量为fraction的数据，withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样，seed用于指定随机数生成器种子。

对源RDD进行去重后返回一个新的RDD。默认情况下，只有8个并行任务来操作，但是可以传入一个可选的numTasks参数改变它

缩减分区数，用于大数据集过滤后，提高小数据集的执行效率

根据分区数，重新通过网络随机洗牌所有数据

coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定

repartition实际上是调用的coalesce，默认是进行shuffle的

使用func先对数据进行处理，按照处理后的数据比较结果排序，默认为正序

并集

差集

交集

笛卡尔积

将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常

对pairRDD进行分区操作，如果原有的partionRDD和现有的partionRDD是一致的话就不进行分区， 否则会生成ShuffleRDD，即会产生shuffle过程

可以通过指定的分区器决定数据计算的分区,spark默认的分区器为HashPartitioner

groupByKey也是对每个key进行操作，但只生成一个sequence

在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置

reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，返回结果是RDD[k,v]

groupByKey：按照key进行分组，直接进行shuffle

一般情况下reduceByKey比groupByKey更好，但要注意combine的使用条件必须是不影响最终的业务逻辑

参数：(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U)

在kv对的RDD中，，按key将value进行分组合并，合并时，将每个value和初始值作为seq函数的参数，进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，最后将每个分组的value传递给combine函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给combine函数，以此类推），将key与计算结果作为一个新的kv对输出。

即计算两次，一次分区内，一次分区外

aggregateByKey分区内和分区外计算规则相同可用foldByKey

(createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C)

createCombiner

mergeValue

mergeCombiners

将第一个key出现的v转换结构计算规则，第二个参数表示分区内计算规则，第三个参数表示分区间计算规则

在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD

针对于(K,V)形式的类型只对V进行操作

在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

.toDebugString

.dependencies

分支主题

Application

Job

Stage

Task

判断相等性的标准方法

hadoopRDD

newHadoopRDD

在Hadoop中以压缩形式存储的数据,不需要指定解压方式就能够进行读取

map-reduce如何读取某一类型数据，将该对应读取方式改写为上述的两种接口

<dependency>

<groupId>mysql</groupId>

<artifactId>mysql-connector-java</artifactId>

<version>5.1.27</version>

</dependency>

MySQL读取

MySQL写入

Spark 可以通过Hadoop输入格式访问HBase

<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>

HBase读取

HBase写入