Spark-Streaming基础
Spark課堂筆記
Spark生態(tài)圈：
Spark Core ： RDD（彈性分布式數(shù)據(jù)集）
Spark SQL
Spark Streaming
Spark MLLib：協(xié)同過濾，ALS，邏輯回歸等等 --> 機(jī)器學(xué)習(xí)
Spark Graphx ： 圖計(jì)算
重點(diǎn)在前三章
-----------------Spark Core------------------------
一、什么是Spark？特點(diǎn)？
https://spark.apache.org/
Apache Spark? is a unified analytics engine for large-scale data processing.
特點(diǎn)：快、易用、通用性、兼容性（完全兼容Hadoop）
快：快100倍（Hadoop 3 之前）
易用：支持多種語言開發(fā)
通用性：生態(tài)系統(tǒng)全。
易用性：兼容Hadoop
spark 取代 Hadoop
二、安裝和部署Spark、Spark 的 HA
1、spark體系結(jié)構(gòu)
Spark的運(yùn)行方式
Yarn
Standalone：本機(jī)調(diào)試（demo）
Worker：從節(jié)點(diǎn)。每個服務(wù)器上，資源和任務(wù)的管理者。只負(fù)責(zé)管理一個節(jié)點(diǎn)。
執(zhí)行過程：
一個Worker 有多個 Executor。 Executor是任務(wù)的執(zhí)行者，按階段（stage）劃分任務(wù)?！?gt; RDD
客戶端：Driver Program 提交任務(wù)到集群中。
1、spark-submit
2、spark-shell
2、spark的搭建
（1）準(zhǔn)備工作：JDK 配置主機(jī)名 免密碼登錄
（2）偽分布式模式
在一臺虛擬機(jī)上模擬分布式環(huán)境（Master和Worker在一個節(jié)點(diǎn)上）
export JAVA_HOME=/usr/java/jdk1.8.0_201
export SPARK_MASTER_HOST=node3
export SPARK_MASTER_PORT=7077
（3）全分布式環(huán)境
修改slave文件 拷貝到其他兩臺服務(wù)器 啟動
3、Spark的 HA
回顧HA；
（*）HDFS Yarn Hbase Spark 主從結(jié)構(gòu)
（*）單點(diǎn)故障
（1）基于文件目錄的單點(diǎn)恢復(fù)
（*）本質(zhì)：還是只有一個主節(jié)點(diǎn)Master，創(chuàng)建了一個恢復(fù)目錄，保存集群狀態(tài)和任務(wù)的信息。
當(dāng)Master掛掉，重新啟動時，會從恢復(fù)目錄下讀取狀態(tài)信息，恢復(fù)出來原來的狀態(tài)
用途：用于開發(fā)和測試，生產(chǎn)用zookeeper
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=FILESYSTEM
-Dspark.deploy.recoveryDirectory=/usr/local/spark-2.1.0-bin-hadoop2.7/recovery"
（2）基于Zookeeper ：和Hadoop類似
（*）復(fù)習(xí)一下zookeeper：
相當(dāng)于一個數(shù)據(jù)庫，把一些信息存放在zookeeper中，比如集群的信息。
數(shù)據(jù)同步功能，選舉功能，分布式鎖功能
數(shù)據(jù)同步：給一個節(jié)點(diǎn)中寫入數(shù)據(jù)，可以同步到其他節(jié)點(diǎn)
選舉：Zookeeper中存在不同的角色，Leader Follower。如果Leader掛掉，重新選舉Leader
分布式鎖：秒殺。以目錄節(jié)點(diǎn)的方式來保存數(shù)據(jù)。
修改 spark-env.sh
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.url=node3:2181,node4:2181,node5:2181
-Dspark.deploy.zookeeper.dir=/spark"
同步到其他兩臺服務(wù)器。
在node3 start-all node3 master node4 Worker node5 Worker
在node4 start-master node3 master node4 master（standby） node4 Worker node5 Worker
在node3上kill master
node4 master（Active） node4 Worker node5 Worker
在網(wǎng)頁http://192.168.109.134:8080/ 可以看到相應(yīng)信息
三、執(zhí)行Spark的任務(wù)：兩個工具
1、spark-submit：用于提交Spark的任務(wù)
任務(wù)：jar。
舉例：蒙特卡洛求PI（圓周率）。
./spark-submit --master spark://node3:7077 --class
--class指明主程序的名字
/usr/local/spark-2.1.0-bin-hadoop2.7/bin/spark-submit --master spark://node3:7077
--class org.apache.spark.examples.SparkPi
/usr/local/spark-2.1.0-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.1.0.jar 100
2、spark-shell 相當(dāng)于REPL
作為一個獨(dú)立的Application運(yùn)行
兩種模式：
（1）本地模式
spark-shell 后面不接任何參數(shù)，代表本地模式
Spark context available as 'sc' (master = local[*], app id = local-1554038459298).
sc 是 SparkContext 對象名。 local[*] 代表本地模式，不提交到集群中運(yùn)行。
（2）集群模式
./spark-submit --master spark://node3:7077 提交到集群中運(yùn)行
Spark context available as 'sc' (master = spark://node3:7077, app id = app-20190331212447-0000).
master = spark://node3:7077
Spark session available as 'spark'
Spark Session 是 2.0 以后提供的，利用 SparkSession 可以訪問spark所有組件。
示例：WordCount程序
（*）處理本地文件，把結(jié)果打印到屏幕上
scala> sc.textFile("/usr/local/tmp_files/test_WordCount.txt")
.flatMap(_.split(" "))
.map((_,1))
.reduceByKey(_+_)
.collect
res0: Array[(String, Int)] = Array((is,1), (love,2), (capital,1), (Beijing,2), (China,2), (I,2), (of,1), (the,1))
（*）處理HDFS文件，結(jié)果保存在hdfs上
sc.textFile("hdfs://node1:8020/tmp_files/test_WordCount.txt")
.flatMap(_.split(" "))
.map((_,1))
.reduceByKey(_+_)
.saveAsTextFile("hdfs://node1:8020/output/0331/test_WordCount")
-rw-r--r-- 3 root supergroup 0 2019-03-31 21:43 /output/0331/test_WordCount/_SUCCESS
-rw-r--r-- 3 root supergroup 40 2019-03-31 21:43 /output/0331/test_WordCount/part-00000
-rw-r--r-- 3 root supergroup 31 2019-03-31 21:43 /output/0331/test_WordCount/part-00001
_SUCCESS 代表程序執(zhí)行成功
part-00000 part-00001 結(jié)果文件，分區(qū)。里面內(nèi)容不重復(fù)。
（*）單步運(yùn)行WordCount ----> RDD
scala> val rdd1 = sc.textFile("/usr/local/tmp_files/test_WordCount.txt")
rdd1: org.apache.spark.rdd.RDD[String] = /usr/local/tmp_files/test_WordCount.txt MapPartitionsRDD[12] at textFile at <console>:24
scala> 1+1
res2: Int = 2
scala> rdd1.collect
res3: Array[String] = Array(I love Beijing, I love China, Beijing is the capital of China)
scala> val rdd2 = rdd1.flatMap(_.split(" "))
rdd2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[13] at flatMap at <console>:26
scala> rdd2.collect
res4: Array[String] = Array(I, love, Beijing, I, love, China, Beijing, is, the, capital, of, China)
scala> val rdd3 = rdd2.map((_,1))
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[14] at map at <console>:28
scala> rdd3.collect
res5: Array[(String, Int)] = Array((I,1), (love,1), (Beijing,1), (I,1), (love,1), (China,1), (Beijing,1), (is,1), (the,1), (capital,1), (of,1), (China,1))
scala> val rdd4 = rdd3.reduceByKey(_+_)
rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[15] at reduceByKey at <console>:30
scala> rdd4.collect
res6: Array[(String, Int)] = Array((is,1), (love,2), (capital,1), (Beijing,2), (China,2), (I,2), (of,1), (the,1))
RDD 彈性分布式數(shù)據(jù)集
（1）依賴關(guān)系 ： 寬依賴和窄依賴
（2）算子：
函數(shù)：
Transformation ： 延時計(jì)算 map flatMap textFile
Action ： 立即觸發(fā)計(jì)算 collect
說明：scala復(fù)習(xí)
（*）flatten：把嵌套的結(jié)果展開
scala> List(List(2,4,6,8,10),List(1,3,5,7,9)).flatten
res21: List[Int] = List(2, 4, 6, 8, 10, 1, 3, 5, 7, 9)
（*）flatmap : 相當(dāng)于一個 map + flatten
scala> var myList = List(List(2,4,6,8,10),List(1,3,5,7,9))
myList: List[List[Int]] = List(List(2, 4, 6, 8, 10), List(1, 3, 5, 7, 9))
scala> myList.flatMap(x=>x.map(_*2))
res22: List[Int] = List(4, 8, 12, 16, 20, 2, 6, 10, 14, 18)
myList.flatMap(x=>x.map(_*2))
執(zhí)行過程：
1、將 List(2, 4, 6, 8, 10), List(1, 3, 5, 7, 9) 調(diào)用 map(_*2) 方法。x 代表一個List
2、flatten
3、在IDE中開發(fā)scala版本和Java版本的WorkCount。
(1)scala版本的WordCount
新建一個工程，把jar引入到工程中。
export jar 點(diǎn)擊下一步下一步，不需要設(shè)置main class
把jar上傳到服務(wù)器上。
spark-submit --master spark://node3:7077
--class day1025.MyWordCount
/usr/local/tmp_files/Demo1.jar
hdfs://node2:8020/tmp_files/test_WordCount.txt
hdfs://node2:8020/output/1025/demo1
（2）java版本的WordCount
./spark-submit --master spark://node3:7077 --class day0330.JavaWordCount /usr/local/tmp_files/Demo2.jar
四、分析Spark的任務(wù)流程
1、分析WordCount程序處理過程
見圖片
2、Spark調(diào)度任務(wù)的過程
提交到及群眾運(yùn)行任務(wù)時，spark執(zhí)行任務(wù)調(diào)度。
見圖片
五、RDD和RDD特性、RDD的算子
1、RDD：彈性分布式數(shù)據(jù)集
（*）Spark中最基本的數(shù)據(jù)抽象。
（*）RDD的特性
* Internally, each RDD is characterized by five main properties:
*
* - A list of partitions
1、是一組分區(qū)。
RDD由分區(qū)組成，每個分區(qū)運(yùn)行在不同的Worker上，通過這種方式來實(shí)現(xiàn)分布式計(jì)算。
* - A function for computing each split
在RDD中，提供算子處理每個分區(qū)中的數(shù)據(jù)
* - A list of dependencies on other RDDs
RDD存在依賴關(guān)系：寬依賴和窄依賴。
* - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
可以自定義分區(qū)規(guī)則來創(chuàng)建RDD
* - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
* an HDFS file)
優(yōu)先選擇離文件位置近的節(jié)點(diǎn)來執(zhí)行
如何創(chuàng)建RDD？
（1）通過SparkContext.parallelize方法來創(chuàng)建
scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[32] at parallelize at <console>:29
scala> rdd1.partitions.length
res35: Int = 3
scala> val rdd1 = sc.parallelize(Array(1,2,3,4,5,6,7,8),2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[33] at parallelize at <console>:29
scala> rdd1.partitions.length
res36: Int = 2
（2）通過外部數(shù)據(jù)源來創(chuàng)建
sc.textFile()
scala> val rdd2 = sc.textFile("/usr/local/tmp_files/test_WordCount.txt")
rdd2: org.apache.spark.rdd.RDD[String] = /usr/local/tmp_files/test_WordCount.txt MapPartitionsRDD[35] at textFile at <console>:29
2、 算子
（1）Transformation
map(func)：相當(dāng)于for循環(huán)，返回一個新的RDD
filter(func)：過濾
flatMap(func)：flat+map 壓平
mapPartitions(func)：對RDD中的每個分區(qū)進(jìn)行操作
mapPartitionsWithIndex(func)：對RDD中的每個分區(qū)進(jìn)行操作，可以取到分區(qū)號。
sample(withReplacement, fraction, seed)：采樣
集合運(yùn)算
union(otherDataset)
intersection(otherDataset)
distinct([numTasks]))：去重
聚合操作：group by
groupByKey([numTasks])
reduceByKey(func, [numTasks])
aggregateByKey(zeroValue)(seqOp,combOp,[numTasks])
排序
sortByKey([ascending], [numTasks])
sortBy(func,[ascending], [numTasks])
join(otherDataset, [numTasks])
cogroup(otherDataset, [numTasks])
cartesian(otherDataset)
pipe(command, [envVars])
coalesce(numPartitions)
重分區(qū)：
repartition(numPartitions)
repartitionAndSortWithinPartitions(partitioner)
舉例：
1、創(chuàng)建一個RDD，每個元素乘以2，再排序
scala> val rdd1 = sc.parallelize(Array(3,4,5,100,79,81,6,8))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[36] at parallelize at <console>:29
scala> val rdd2 = rdd1.map(_*2)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[37] at map at <console>:31
scala> rdd2.collect
res37: Array[Int] = Array(6, 8, 10, 200, 158, 162, 12, 16)
scala> rdd2.sortBy(x=>x,true).collect
res39: Array[Int] = Array(6, 8, 10, 12, 16, 158, 162, 200)
scala> rdd2.sortBy(x=>x,false).collect
res40: Array[Int] = Array(200, 162, 158, 16, 12, 10, 8, 6)
def sortBy[K](f: (T) ? K, ascending: Boolean = true)
過濾出大于20的元素：
scala> val rdd3 = rdd2.filter(_>20)
rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[53] at filter at <console>:33
scala> rdd3.collect
res41: Array[Int] = Array(200, 158, 162)
2、字符串（字符）類型的RDD
scala> val rdd4 = sc.parallelize(Array("a b c","d e f","g h i"))
rdd4: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[54] at parallelize at <console>:29
scala> rdd4.flatMap(_.split(" ")).collect
res42: Array[String] = Array(a, b, c, d, e, f, g, h, i)
3、RDD的集合運(yùn)算：
scala> val rdd6 = sc.parallelize(List(1,2,3,6,7,8,9,100))
rdd6: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[56] at parallelize at <console>:29
scala> val rdd7 = sc.parallelize(List(1,2,3,4))
rdd7: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[57] at parallelize at <console>:29
scala> val rdd8 = rdd6.union(rdd7)
rdd8: org.apache.spark.rdd.RDD[Int] = UnionRDD[58] at union at <console>:33
scala> rdd8.collect
res43: Array[Int] = Array(1, 2, 3, 6, 7, 8, 9, 100, 1, 2, 3, 4)
scala> rdd8.distinct.collect
res44: Array[Int] = Array(100, 4, 8, 1, 9, 6, 2, 3, 7)
4、分組操作：reduceByKey
<key value>
scala> val rdd1 = sc.parallelize(List(("Tom",1000),("Andy",2000),("Lily",1500)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[62] at parallelize at <console>:29
scala> val rdd2 = sc.parallelize(List(("Andy",1000),("Tom",2000),("Mike",500)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[63] at parallelize at <console>:29
scala> val rdd3 = rdd1 union rdd2
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[64] at union at <console>:33
scala> rdd3.collect
res45: Array[(String, Int)] = Array((Tom,1000), (Andy,2000), (Lily,1500), (Andy,1000), (Tom,2000), (Mike,500))
scala> val rdd4= rdd3.groupByKey
rdd4: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = ShuffledRDD[65] at groupByKey at <console>:35
scala> rdd4.collect
res46: Array[(String, Iterable[Int])] = Array(
(Tom,CompactBuffer(1000, 2000)),
(Andy,CompactBuffer(2000, 1000)),
(Mike,CompactBuffer(500)), (
Lily,CompactBuffer(1500)))
scala> rdd3.reduceByKey(_+_).collect
res47: Array[(String, Int)] = Array((Tom,3000), (Andy,3000), (Mike,500), (Lily,1500))
reduceByKey will provide much better performance.
官方不推薦使用 groupByKey 推薦使用 reduceByKey
5、cogroup
scala> val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[67] at parallelize at <console>:29
scala> val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[68] at parallelize at <console>:29
scala> val rdd3 = rdd1.cogroup(rdd2)
rdd3: org.apache.spark.rdd.RDD[(String, (Iterable[Int], Iterable[Int]))] = MapPartitionsRDD[70] at cogroup at <console>:33
scala> rdd3.collect
res48: Array[(String, (Iterable[Int], Iterable[Int]))] = Array(
(tom,(CompactBuffer(1, 2),CompactBuffer(1))),
(jerry,(CompactBuffer(3),CompactBuffer(2))),
(shuke,(CompactBuffer(),CompactBuffer(2))),
(kitty,(CompactBuffer(2),CompactBuffer())))
6、reduce操作（Action）
聚合操作
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[71] at parallelize at <console>:29
scala> rdd1.reduce(_+_)
res49: Int = 15
7、需求：按照value排序。
做法：
1、交換，把key 和 value交換，然后調(diào)用sortByKey方法
2、再次交換
scala> val rdd1 = sc.parallelize(List(("tom",1),("jerry",3),("ketty",2),("shuke",2)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[72] at parallelize at <console>:29
scala> val rdd2 = sc.parallelize(List(("jerry",1),("tom",3),("shuke",5),("ketty",1)))
rdd2: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[73] at parallelize at <console>:29
scala> val rdd3 = rdd1.union(rdd2)
rdd3: org.apache.spark.rdd.RDD[(String, Int)] = UnionRDD[74] at union at <console>:33
scala> val rdd4 = rdd3.reduceByKey(_+_)
rdd4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[75] at reduceByKey at <console>:35
scala> rdd4.collect
res50: Array[(String, Int)] = Array((tom,4), (jerry,4), (shuke,7), (ketty,3))
scala> val rdd5 = rdd4.map(t=>(t._2,t._1)).sortByKey(false).map(t=>(t._2,t._1))
rdd5: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[80] at map at <console>:37
scala> rdd5.collect
res51: Array[(String, Int)] = Array((shuke,7), (tom,4), (jerry,4), (ketty,3))
（2）Action
reduce(func)
collect()
count()
first()
take(n)
takeSample(withReplacement,num, [seed])
takeOrdered(n,?[ordering])
saveAsTextFile(path)
saveAsSequenceFile(path)?
saveAsObjectFile(path)?
countByKey()
foreach(func)：與map類似，沒有返回值。
3、特性：
（1）RDD的緩存機(jī)制
（*）作用：提高性能
（*）使用：標(biāo)識RDD可以被緩存 persist cache
（*）可以緩存的位置：
val NONE = new StorageLevel(false, false, false, false)
val DISK_ONLY = new StorageLevel(true, false, false, false)
val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
val MEMORY_ONLY = new StorageLevel(false, true, false, true)
val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
/**
* Persist this RDD with the default storage level (`MEMORY_ONLY`).
*/
def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)
/**
* Persist this RDD with the default storage level (`MEMORY_ONLY`).
*/
def cache(): this.type = persist()
舉例：測試數(shù)據(jù)，92萬條
scala> val rdd1 = sc.textFile("hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt")
rdd1: org.apache.spark.rdd.RDD[String] = hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt MapPartitionsRDD[82] at textFile at <console>:29
scala> rdd1.count --> 直接出發(fā)計(jì)算
res52: Long = 923452
scala> rdd1.cache --> 標(biāo)識RDD可以被緩存，不會觸發(fā)計(jì)算
res53: rdd1.type = hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt MapPartitionsRDD[82] at textFile at <console>:29
scala> rdd1.count --> 和第一步一樣，觸發(fā)計(jì)算，但是，把結(jié)果進(jìn)行緩存
res54: Long = 923452
scala> rdd1.count --> 從緩存中直接讀出結(jié)果
res55: Long = 923452
（2）RDD的容錯機(jī)制：通過檢查點(diǎn)來實(shí)現(xiàn)。
/**
* Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint
* directory set with `SparkContext#setCheckpointDir` and all references to its parent
* RDDs will be removed. This function must be called before any job has been
* executed on this RDD. It is strongly recommended that this RDD is persisted in
* memory, otherwise saving it on a file will require recomputation.
*/
（*）復(fù)習(xí)檢查點(diǎn)：HDFS中的檢查點(diǎn)：有SecondaryNamenode來實(shí)現(xiàn)日志的合并。
（*）RDD的檢查點(diǎn)：容錯
概念：血統(tǒng) Lineage
理解：表示任務(wù)執(zhí)行的生命周期。
WordCount textFile ---> redceByKey
如果血統(tǒng)越長，越容易出錯。
假如有檢查點(diǎn)，可以從最近的一個檢查點(diǎn)開始，往后面計(jì)算。不用重頭計(jì)算。
（*）RDD檢查點(diǎn)的類型：
（1）基于本地目錄：需要將Spark shell 或者任務(wù)運(yùn)行在本地模式上（setMaster("local")）
開發(fā)和測試
（2）HDFS目錄：用于生產(chǎn)。
sc.setCheckPointDir(目錄)
舉例：設(shè)置檢查點(diǎn)
scala> var rdd1 = sc.textFile("hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt")
rdd1: org.apache.spark.rdd.RDD[String] = hdfs://192.168.109.131:8020/tmp_files/test_Cache.txt MapPartitionsRDD[1] at textFile at <console>:24
設(shè)置檢查點(diǎn)目錄：
scala> sc.setCheckpointDir("hdfs://192.168.109.131:8020/sparkckpt")
標(biāo)識rdd1可以執(zhí)行檢查點(diǎn)操作
scala> rdd1.checkpoint
scala> rdd1.count
res2: Long = 923452
（3）依賴關(guān)系：寬依賴，窄依賴。
劃分任務(wù)執(zhí)行的stage
見講義。
六、RDD的高級算子
1、mapPartitionsWithIndex：對RDD中的每個分區(qū)（帶有下標(biāo)）進(jìn)行操作，下標(biāo)用index表示
通過這個算子，我們可以獲取分區(qū)號。
def mapPartitionsWithIndex[U](
f: (Int, Iterator[T]) ? Iterator[U],
preservesPartitioning: Boolean = false)(
implicit arg0: ClassTag[U]): RDD[U]
參數(shù)：f是個函數(shù)參數(shù) f 中第一個參數(shù)是Int，代表分區(qū)號，第二個Iterator[T]代表分區(qū)中的元素
舉例：把分區(qū)中的元素，包括分區(qū)號，都打印出來。
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8),3)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at parallelize at <console>:24
scala> def fun1(index:Int, iter:Iterator[Int]) : Iterator[String] = {
| iter.toList.map(x => "[partId : " + index + " , value = " + x + " ]").iterator
| }
fun1: (index: Int, iter: Iterator[Int])Iterator[String]
scala> rdd1.mapPartitions
mapPartitions mapPartitionsWithIndex
scala> rdd1.mapPartitionsWithIndex(fun1).collect
res3: Array[String] = Array(
[partId : 0 , value = 1 ], [partId : 0 , value = 2 ],
[partId : 1 , value = 3 ], [partId : 1 , value = 4 ], [partId : 1 , value = 5 ],
[partId : 2 , value = 6 ], [partId : 2 , value = 7 ], [partId : 2 , value = 8 ])
2、aggregate：聚合操作。類似于分組。
（*）先對局部進(jìn)行聚合操作，再對全局進(jìn)行聚合操作。
調(diào)用聚合操作
scala> val rdd2 = sc.parallelize(List(1,2,3,4,5),2)
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at parallelize at <console>:24
scala> rdd2.mapPartitionsWithIndex(fun1).collect
res4: Array[String] = Array(
[partId : 0 , value = 1 ], [partId : 0 , value = 2 ],
[partId : 1 , value = 3 ], [partId : 1 , value = 4 ], [partId : 1 , value = 5 ])
scala> import scala.math._
import scala.math._
scala> rdd2.aggregate(0)(max(_,_),_+_)
res6: Int = 7
說明：aggregate
(0) 初始值是 0
(max(_,_) 局部操作的函數(shù)
, _+_ 全局操作的函數(shù)
)
scala> rdd2.aggregate(100)(max(_,_),_+_)
res8: Int = 300
分析結(jié)果：初始值是100，代表每個分區(qū)多了一個100
全局操作，也多了一個100
100+100+100 = 300
對RDD中的元素進(jìn)行求和
1、RDD.map
2、聚合操作
scala> rdd2.aggregate(0)(_+_,_+_)
res9: Int = 15
MapReduce Combiner
scala> rdd2.aggregate(10)(_+_,_+_)
res10: Int = 45
（*）對字符串操作
scala> val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[7] at parallelize at <console>:27
scala> rdd2.aggregate("")(_+_,_+_)
res11: String = abcdef
scala> rdd2.aggregate("*")(_+_,_+_)
res12: String = **def*abc
結(jié)果分析:
1、*abc *def
2、**def*abc
（*）復(fù)雜的例子：
1、
scala> val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[8] at parallelize at <console>:27
scala> def fun1(index:Int, iter:Iterator[String]) : Iterator[String] = {
| iter.toList.map(x => "[partId : " + index + " , value = " + x + " ]").iterator
| }
scala> rdd3.mapPartitionsWithIndex(fun1).collect
res17: Array[String] = Array(
[partId : 0 , value = 12 ], [partId : 0 , value = 23 ],
[partId : 1 , value = 345 ], [partId : 1 , value = 4567 ])
scala> rdd3.aggregate("")((x,y)=> math.max(x.length,y.length).toString,(x,y)=>x+y)
res13: String = 42
執(zhí)行過程：
第一個分區(qū)：
第一次比較： "" "12" 長度最大值 2 2-->"2"
第二次比較： “2” “23” 長度最大值 2 2-->"2"
第二個分區(qū)：
第一次比較： "" "345" 長度最大值 3 3-->"3"
第二次比較： “3” “4567” 長度最大值 4 4-->"4"
2、
rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
scala> rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
res18: String = 11
執(zhí)行過程：
第一個分區(qū)：
第一次比較： "" "12" 長度最小值 0 0-->"0"
第二次比較： “0” “23” 長度最小值 1 1-->"1"
第二個分區(qū)：
第一次比較： "" "345" 長度最小值 0 0-->"0"
第二次比較： “0” “4567” 長度最小值 1 1-->"1"
val rdd3 = sc.parallelize(List("12","23","345",""),2)
rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
scala> val rdd3 = sc.parallelize(List("12","23","345",""),2)
rdd3: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[10] at parallelize at <console>:27
scala> rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
res19: String = 10
scala> rdd3.aggregate("")((x,y)=> math.min(x.length,y.length).toString,(x,y)=>x+y)
res20: String = 01
3、aggregateByKey：類似于aggregate，區(qū)別：操作的是 key value 的數(shù)據(jù)類型。
scala> def fun3(index:Int, iter:Iterator[(String,Int)]) : Iterator[String] = {
| iter.toList.map(x => "[partId : " + index + " , value = " + x + " ]").iterator
| }
scala> pairRDD.mapPartitionsWithIndex(fun3).collect
res22: Array[String] = Array(
[partId : 0 , value = (cat,2) ], [partId : 0 , value = (cat,5) ], [partId : 0 , value = (mouse,4) ],
[partId : 1 , value = (cat,12) ], [partId : 1 , value = (dog,12) ], [partId : 1 , value = (mouse,2) ])
（1）將每個動物園（分區(qū)）中，動物數(shù)最多的動物，進(jìn)行求和
動物園0
[partId : 0 , value = (cat,2) ], [partId : 0 , value = (cat,5) ], [partId : 0 , value = (mouse,4) ],
動物園1
[partId : 1 , value = (cat,12) ], [partId : 1 , value = (dog,12) ], [partId : 1 , value = (mouse,2) ])
pairRDD.aggregateByKey(0)(math.max(_,_),_+_)
scala> pairRDD.aggregateByKey(0)(math.max(_,_),_+_).collect
res24: Array[(String, Int)] = Array((dog,12), (cat,17), (mouse,6))
（2）將所有動物求和
pairRDD.aggregateByKey(0)(_+_,_+_).collect
scala> pairRDD.reduceByKey(_+_).collect
res27: Array[(String, Int)] = Array((dog,12), (cat,19), (mouse,6))
aggregateByKey效率更高。
4、 coalesce與repartition
與分區(qū)有關(guān)
都是對RDD進(jìn)行重分區(qū)。
區(qū)別：
coalesce 默認(rèn)不會進(jìn)行Shuffle 默認(rèn) false 如需修改分區(qū)，需置為true
repartition 會進(jìn)行Shuffle
scala> val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),2)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[17] at parallelize at <console>:27
scala> val rdd2 = rdd1.repartition(3)
rdd2: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[21] at repartition at <console>:29
scala> rdd2.partitions.length
res28: Int = 3
scala> val rdd3 = rdd1.coalescse(3,true)
<console>:29: error: value coalescse is not a member of org.apache.spark.rdd.RDD[Int]
val rdd3 = rdd1.coalescse(3,true)
^
scala> val rdd3 = rdd1.coalesce(3,true)
rdd3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[25] at coalesce at <console>:29
scala> rdd3.partitions.length
res29: Int = 3
scala> val rdd4 = rdd1.coalesce(4)
rdd4: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[26] at coalesce at <console>:29
scala> rdd4.partitions.length
res30: Int = 2
5、其他高級算子
http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html
七、編程案例
（1）分析日志
需求：找到訪問量最高的兩個網(wǎng)頁
（*）第一步：對網(wǎng)頁的訪問量求和
（*）第二步：排序，降序
（2）創(chuàng)建自定義分區(qū)
（3）使用JDBCRDD 操作數(shù)據(jù)庫
（4）操作數(shù)據(jù)庫：把結(jié)果存放到數(shù)據(jù)庫中
IDE
-----------------Spark SQL------------------------
類似于Hive
一、Spark SQL 基礎(chǔ)
1、什么是Spark SQL
Spark SQL is Apache Spark's module for working with structured data.
Spark SQL 是spark 的一個模塊。來處理 結(jié)構(gòu)化 的數(shù)據(jù)
不能處理非結(jié)構(gòu)化的數(shù)據(jù)
特點(diǎn)：
1、容易集成
不需要單獨(dú)安裝。
2、統(tǒng)一的數(shù)據(jù)訪問方式
結(jié)構(gòu)化數(shù)據(jù)的類型：JDBC JSon Hive parquer文件 都可以作為Spark SQL 的數(shù)據(jù)源
對接多種數(shù)據(jù)源，且使用方式類似
3、完全兼容hive
把Hive中的數(shù)據(jù)，讀取到Spark SQL中運(yùn)行。
4、支持標(biāo)準(zhǔn)的數(shù)據(jù)連接
JDBC
2、為什么學(xué)習(xí)Spark SQL
執(zhí)行效率比Hive高
hive 2.x 執(zhí)行引擎可以使用 Spark
3、核心概念：表（DataFrame DataSet）
mysql中的表：表結(jié)構(gòu)、數(shù)據(jù)
DataFrame：Schema、RDD（數(shù)據(jù)）
DataSet 在spark1.6以后，對DataFrame做了一個封裝。
4、創(chuàng)建DataFrame
（*）測試數(shù)據(jù)：員工表、部門表
第一種方式：使用case class
1、定義Schema
樣本類來定義Schema。
case class 特點(diǎn)：
可以支持模式匹配，使用case class建立表結(jié)構(gòu)
7521, WARD, SALESMAN,7698, 1981/2/22, 1250, 500, 30
case class Emp(empno:Int,ename:String,job:String,mgr:Int,hiredate:String,sal:Int,comm:Int,deptno:Int)
2、讀取文件
val lines = sc.textFile("/usr/local/tmp_files/emp.csv").map(_.split(","))
3、把每行數(shù)據(jù)，映射到Emp上。
val allEmp = lines.map(x => Emp(x(0).toInt,x(1),x(2),x(3).toInt,x(4),x(5).toInt,x(6).toInt,x(7).toInt))
4、生成DataFrame
val df1 = allEmp.toDF
df1.show
第二種方式 使用Spark Session
（1）什么是Spark Session
Spark session available as 'spark'.
2.0以后引入的統(tǒng)一訪問方式。可以訪問所有的Spark組件。
def createDataFrame(rowRDD: RDD[Row], schema: StructType): DataFrame
（2）使用StructType來創(chuàng)建Schema
val struct =
StructType(
StructField("a", IntegerType, true) ::
StructField("b", LongType, false) ::
StructField("c", BooleanType, false) :: Nil)
case class Emp(empno:Int,
ename:String,
job:String,
mgr:Int,
hiredate:String,
sal:Int,
comm:Int,
deptno:Int)
val myschema = StructType(
List(
StructField("empno",DataTypes.IntegerType),
StructField("ename",DataTypes.StringType),
StructField("job",DataTypes.StringType),
StructField("mgr",DataTypes.IntegerType),
StructField("hiredate",DataTypes.StringType),
StructField("sal",DataTypes.IntegerType),
StructField("comm",DataTypes.IntegerType),
StructField("deptno",DataTypes.IntegerType),
))
import org.apache.spark.sql.types._
準(zhǔn)備數(shù)據(jù) RDD[Row]
val allEmp = lines.map(x => Row(x(0).toInt,x(1),x(2),x(3).toInt,x(4),x(5).toInt,x(6).toInt,x(7).toInt))
import org.apache.spark.sql.Row
val df2 = spark.createDataFrame(allEmp,myschema)
第三種方式
直接讀取一個帶格式的文件。
val df3 = spark.read 讀文件，默認(rèn)是Parquet文件
val df3 = spark.read.json("/usr/local/tmp_files/people.json")
df3.show
val df4 = spark.read.format("json").load("/usr/local/tmp_files/people.json")
5、操作DataFrame
（1）DSL語句
mybatis Hibernate
df1.select($"ename",$"sal",$"sal"+100).show
$"sal" 可以看做是一個變量。
查詢薪水大于2000的員工
df1.filter($"sal" > 2000).show
求每個部門的員工人數(shù)
df1.groupBy($"deptno").count.show
select deptno,count(1) from emp group by deptno
（2）SQL語句
注意：不能直接執(zhí)行SQL，需要生成一個視圖，再執(zhí)行sql。
scala> df1.create
createGlobalTempView createOrReplaceTempView createTempView
一般用到 createOrReplaceTempView createTempView
視圖：類似于表，但不保存數(shù)據(jù)。
df1.createOrReplaceTempView("emp")
操作：
spark.sql("select * from emp").show
查詢薪水大于2000的員工
spark.sql("select * from emp where sal > 2000").show
求每個部門的員工人數(shù)
spark.sql("select deptno,count(1) from emp group by deptno").show
（3）多表查詢
10,ACCOUNTING,NEW YORK
case class Dept(deptno:Int,dname:String,loc:String)
val lines = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))
val allDept = lines.map(x=>Dept(x(0).toInt,x(1),x(2)))
df5.createOrReplaceTempView("dept")
spark.sql("select dname,ename from emp,dept where emp.deptno=dept.deptno").show
6、操作DataSet
跟DataFrame類似，是一套新的接口。高級的Dataframe
舉例：
（1）創(chuàng)建DataSet
1、使用序列來創(chuàng)建DataSet。
定義一個case class
case class MyData(a:Int,b:String)
生成序列，并創(chuàng)建DataSet
val ds = Seq(MyData(1,"Tom"),MyData(2,"Mary")).toDS
.toDS 生成DataSet
ds.show
2、使用JSON數(shù)據(jù)來創(chuàng)建DataSet
定義case class
case class Person(name:String,age:BigInt)
通過Json數(shù)據(jù)來生成DataFrame
val df = spark.read.format("json").load("/usr/local/tmp_files/people.json")
將DataFrame轉(zhuǎn)換成DataSet
df.as[Person].show
df.as[Person] 就是一個DataSet
3、使用其他數(shù)據(jù)
RDD操作和DataFrame操作相結(jié)合 ---> DataSet
讀取數(shù)據(jù)，創(chuàng)建DataSet
val linesDS = spark.read.text("/usr/local/tmp_files/test_WordCount.txt").as[String]
對DataSet進(jìn)行操作：
val words = linesDS.flatMap(_.split(" ")).filter(_.length > 3)
words.show
words.collect
執(zhí)行一個WordCount程序
val result = linesDS.flatMap(_.split(" ")).map((_,1)).groupByKey( x => x._1).count
result.show
排序：
result.orderBy($"value").show
result.orderBy($"count(1)").show
（2）DataSet操作案例
使用emp.json 生成一個DataFrame
val empDF = spark.read.json("/usr/local/tmp_files/emp.json")
查詢工資大于3000的員工
empDF.where($"sal" >= 3000).show
創(chuàng)建case class
case class Emp(empno:BigInt,ename:String,job:String,mgr:String,hiredate:String,sal:BigInt,comm:String,deptno:BigInt)
生成DataSet
val empDS = empDF.as[Emp]
查詢工資大于3000的員工
empDS.filter(_.sal > 3000).show
查詢10號部門的員工
empDS.filter(_.deptno == 10).show
（3）多表查詢
1、創(chuàng)建部門表
val deptRDD = sc.textFile("/usr/local/tmp_files/dept.csv").map(_.split(","))
case class Dept(deptno:Int,dname:String,loc:String)
val deptDS = deptRDD.map( x=> Dept(x(0).toInt,x(1),x(2))).toDS
2、創(chuàng)建員工表
case class Emp(empno:Int,ename:String,job:String,mgr:Int,hiredate:String,sal:Int,comm:Int,deptno:Int)
val empRDD = sc.textFile("/usr/local/tmp_files/emp.csv").map(_.split(","))
7369,SMITH,CLERK,7902,1980/12/17,800,0,20
val empDS = empRDD.map(x=> Emp(x(0).toInt,x(1),x(2),x(3).toInt,x(4),x(5).toInt,x(6).toInt,x(7).toInt)).toDS
3、執(zhí)行多表查詢：等值連接
val result = deptDS.join(empDS,"deptno")
val result1 = deptDS.joinWith(empDS, deptDS("deptno") === empDS("deptno") )
join 和 joinWith 區(qū)別：連接后schema不同
4、多表連接后再篩選
deptDS.join(empDS,"deptno").where("deptno == 10").show
7、Spark SQL 中的視圖
視圖是一個虛表，不存儲數(shù)據(jù)。
兩種類型：
1、普通視圖（本地視圖）：只在當(dāng)前Session中有效。createOrReplaceTempView createTempView
2、全局視圖： createGlobalTempView
在不同的Session中都有用 把全局視圖創(chuàng)建在命名空間中：global_temp中。類似于一個庫。
scala> df1.create
createGlobalTempView createOrReplaceTempView createTempView
舉例：
創(chuàng)建一個新session，讀取不到emp視圖
spark.newSession.sql("select * from emp")
以下兩種方式均可讀到 全局視圖 中的數(shù)據(jù)。
df1.createGlobalTempView("emp1")
spark.newSession.sql("select * from global_temp.emp1").show
spark.sql("select * from global_temp.emp1").show
二、使用數(shù)據(jù)源
在Spark SQL中，可以使用各種各樣的數(shù)據(jù)源來操作。 結(jié)構(gòu)化
1、使用load函數(shù)、save函數(shù)
load函數(shù)是加載數(shù)據(jù)，save是存儲數(shù)據(jù)。
注意：使用load 或 save時，默認(rèn)是Parquet文件。列式存儲文件。
舉例:
讀取 users.parquet 文件
val userDF = spark.read.load("/usr/local/tmp_files/users.parquet")
userDF.printSchema
userDF.show
val userDF = spark.read.load("/usr/local/tmp_files/emp.json")
保存parquet文件
userDF.select($"name",$"favorite_color").write.save("/usr/local/tmp_files/parquet")
讀取剛剛寫入的文件：
val userDF1 = spark.read.load("/usr/local/tmp_files/parquet/part-00000-1ab4e661-32c6-441a-b320-79d")---> 不推薦
生產(chǎn)：
val userDF2 = spark.read.load("/usr/local/tmp_files/parquet")
讀json文件 必須format
val userDF = spark.read.format("json").load("/usr/local/tmp_files/emp.json")
val userDF3 = spark.read.json("/usr/local/tmp_files/emp.json")
關(guān)于save函數(shù)：
調(diào)用save函數(shù)的時候，可以指定存儲模式，追加、覆蓋等等
userDF2.write.save("/usr/local/tmp_files/parquet")
userDF2.write.save("/usr/local/tmp_files/parquet")
org.apache.spark.sql.AnalysisException: path file:/usr/local/tmp_files/parquet already exists.;
save的時候覆蓋
userDF2.write.mode("overwrite").save("/usr/local/tmp_files/parquet")
將結(jié)果保存成表
userDF2.select($"name").write.saveAsTable("table1")
scala> userDF.select($"name").write.saveAsTable("table2")
scala> spark.sql("select * from table2").show
+------+
| name|
+------+
|Alyssa|
| Ben|
+------+
2、Parquet文件：列式存儲文件，是Spark SQL 默認(rèn)的數(shù)據(jù)源
就是一個普通的文件
舉例：
1、把其他文件，轉(zhuǎn)換成Parquet文件
調(diào)用save函數(shù)
把數(shù)據(jù)讀進(jìn)來，再寫出去，就是Parquet文件。
val empDF = spark.read.json("/usr/local/tmp_files/emp.json")
empDF.write.mode("overwrite").save("/usr/local/tmp_files/parquet")
empDF.write.mode("overwrite").parquet("/usr/local/tmp_files/parquet")
val emp1 = spark.read.parquet("/usr/local/tmp_files/parquet")
emp1.createOrReplaceTempView("emp1")
spark.sql("select * from emp1")
2、支持Schema的合并
項(xiàng)目開始 表結(jié)構(gòu)簡單 schema簡單
項(xiàng)目越來越大 schema越來越復(fù)雜
舉例：
通過RDD來創(chuàng)建DataFrame
val df1 = sc.makeRDD(1 to 5).map( i => (i,i*2)).toDF("single","double")
"single","double" 是表結(jié)構(gòu)
df1.show
df1.write.mode("overwrite").save("/usr/local/tmp_files/test_table/key=1")
val df2 = sc.makeRDD(6 to 10).map( i => (i,i*3)).toDF("single","triple")
df2.show
df2.write.mode("overwrite").save("/usr/local/tmp_files/test_table/key=2")
合并兩個部分
val df3 = spark.read.parquet("/usr/local/tmp_files/test_table")
val df3 = spark.read.option("mergeSchema",true).parquet("/usr/local/tmp_files/test_table")
通過RDD來創(chuàng)建DataFrame
val df1 = sc.makeRDD(1 to 5).map( i => (i,i*2)).toDF("single","double")
"single","double" 是表結(jié)構(gòu)
df1.show
df1.write.mode("overwrite").save("/usr/local/tmp_files/test_table/tzkt=1")
val df2 = sc.makeRDD(6 to 10).map( i => (i,i*3)).toDF("single","triple")
df2.show
df2.write.mode("overwrite").save("/usr/local/tmp_files/test_table/key=2")
合并兩個部分
val df3 = spark.read.parquet("/usr/local/tmp_files/test_table")
val df3 = spark.read.option("mergeSchema",true).parquet("/usr/local/tmp_files/test_table")
3、json文件
讀取Json文件，生成DataFrame
val peopleDF = spark.read.json("/usr/local/tmp_files/people.json")
peopleDF.printSchema
peopleDF.createOrReplaceTempView("peopleView")
spark.sql("select * from peopleView").show
Spark SQL 支持統(tǒng)一的訪問接口。對于不同的數(shù)據(jù)源，讀取進(jìn)來，生成DataFrame后，操作完全一樣。
4、JDBC
使用JDBC操作關(guān)系型數(shù)據(jù)庫，加載到Spark中進(jìn)行分析和處理。
方式一：
val mysqlDF = spark.read.format("jdbc")
.option("url","jdbc:mysql://192.168.109.1:3306/company?serverTimezone=UTC&characterEncoding=utf-8")
.option("user","root")
.option("password","123456")
.option("driver","com.mysql.jdbc.Driver")
.option("dbtable","emp").load
mysqlDF.show
方式二：
定義一個Properties類
import java.util.Properties
val mysqlProps = new Properties()
mysqlProps.setProperty("user","root")
mysqlProps.setProperty("password","123456")
val mysqlDF1 = spark.read.jdbc("jdbc:mysql://192.168.109.1:3306/company?serverTimezone=UTC&characterEncoding=utf-8","emp",mysqlProps)
mysqlDF1.show
5、使用Hive
比較常見
（*）spark SQL 完全兼容hive
（*）需要進(jìn)行配置
拷貝一下文件到spark/conf目錄下：
Hive 配置文件： hive-site.xml
Hadoop 配置文件：core-site.xml hdfs-site.xml
配置好后，重啟spark
啟動Hadoop 與 hive
spark.sql("create table comany.emp_0410(empno Int,ename String,job String,mgr String,hiredate String,sal Int,comm String,deptno Int)row format delimited fields terminated by ','")
三、在IDE中開發(fā)Spark SQL
四、性能優(yōu)化
與RDD類似
1、把內(nèi)存中緩存表的數(shù)據(jù)
直接讀取內(nèi)存的值，來提高性能。
RDD中如何緩存：
rdd.cache 或者 rdd.persist
在Spark SQL中，使用SparkSession.sqlContext.cacheTable
spark中所有context對象
1、sparkContext ： SparkCore
2、sql Context ： SparkSQL
3、Streaming Context ：SparkStreaming
統(tǒng)一起來：SparkSession
操作mysql，啟動spark shell 時，需要：
./bin/spark-shell --master spark://node3:7077 --jars /usr/local/tmp_files/mysql-connector-java-8.0.11.jar --driver-class-path /usr/local/tmp_files/mysql-connector-java-8.0.11.jar
val mysqlDF = spark.read.format("jdbc").option("driver","com.mysql.jdbc.Driver").option("url","jdbc:mysql://192.168.109.1:3306/company?serverTimezone=UTC&characterEncoding=utf-8").option("user","root").option("password","123456").option("dbtable","emp").load
mysqlDF.show
mysqlDF.createOrReplaceTempView("emp")
spark.sqlContext.cacheTable("emp") ----> 標(biāo)識這張表可以被緩存，數(shù)據(jù)還沒有真正被緩存
spark.sql("select * from emp").show ----> 依然讀取mysql
spark.sql("select * from emp").show ----> 從緩存中讀取數(shù)據(jù)
spark.sqlContext.clearCache
清空緩存后，執(zhí)行查詢，會觸發(fā)查詢mysql數(shù)據(jù)庫。
2、了解性能優(yōu)化的相關(guān)參數(shù)：參考講義
-----------------Spark Streaming------------------------
流式計(jì)算框架，類似于Storm
常用的實(shí)時計(jì)算引擎（流式計(jì)算）
1、Apache Storm：真正的流式計(jì)算
2、Spark Streaming ：嚴(yán)格上來說，不是真正的流式計(jì)算（實(shí)時計(jì)算）
把連續(xù)的流式數(shù)據(jù)，當(dāng)成不連續(xù)的RDD
本質(zhì)：是一個離散計(jì)算（不連續(xù)）
3、Apache Flink：真正的流式計(jì)算。與Spark Streaming相反。
把離散的數(shù)據(jù)，當(dāng)成流式數(shù)據(jù)來處理
4、JStorm
一、Spark Streaming基礎(chǔ)
1、什么是 Spark Streaming。
Spark Streaming makes it easy to build scalable fault-tolerant streaming applications.
易于構(gòu)建靈活的、高容錯的流式系統(tǒng)。
特點(diǎn)：
1、易用，已經(jīng)集成到Spark中
2、容錯性：底層RDD，RDD本身具有容錯機(jī)制
3、支持多種語言：Java Scala Python
2、演示官方的Demo
往Spark Streaming中發(fā)送字符串，Spark 接收到以后，進(jìn)行計(jì)數(shù)
使用消息服務(wù)器 netcat Linux自帶
yum install nc.x86_64
nc -l 1234
注意：總核心數(shù) 大于等于2。一個核心用于接收數(shù)據(jù)，另一個用于處理數(shù)據(jù)
在netcat中寫入數(shù)據(jù) Spark Streaming可以取到
3、開發(fā)自己的NetWorkWordCount程序
和Spark Core類似
問題：Hello Hello
Hello World
現(xiàn)在現(xiàn)象：（Hello,2）
(Hello , 1) (World , 1)
能不能累加起來？保存記錄下以前的狀態(tài)？
通過Spark Streaming提供的算子來實(shí)現(xiàn)
二、高級特性
1、什么是DStream？離散流
把連續(xù)的數(shù)據(jù)變成不連續(xù)的RDD
因?yàn)镈Stream的特性，導(dǎo)致，Spark Streaming不是真正的流式計(jì)算
2、重點(diǎn)算子講解
（1）updateStateByKey
默認(rèn)情況下，Spark Streaming不記錄之前的狀態(tài)，每次發(fā)數(shù)據(jù)，都會從0開始
現(xiàn)在使用本算子，實(shí)現(xiàn)累加操作。
（2）transform
3、窗口操作
窗口：對落在窗口內(nèi)的數(shù)據(jù)進(jìn)行處理，也是一個DStream，RDD
舉例：每10秒鐘把過去30秒的數(shù)據(jù)采集過來
注意：先啟動nc 再啟動程序 local[2]
4、集成Spark SQL ： 使用SQL語句來處理流式數(shù)據(jù)
5、緩存和持久化：和RDD一樣
6、支持檢查點(diǎn)：和RDD一樣
三、數(shù)據(jù)源
Spark Streaming是一個流式計(jì)算引擎，就需要從外部數(shù)據(jù)源來接收數(shù)據(jù)
1、基本的數(shù)據(jù)源
文件流：監(jiān)控文件系統(tǒng)的變化，如果文件有增加，讀取文件中的內(nèi)容
希望Spark Streaming監(jiān)控一個文件夾，如果有變化，則把變化采集過來
RDD隊(duì)列流：可以從隊(duì)列中獲取數(shù)據(jù)
套接字流：socketTextStream
2、高級數(shù)據(jù)源
（1）Flume
Spark SQL 對接flume有多種方式：
push方式：flume將數(shù)據(jù)推送給Spark Streaming
（2）Kafka
在講Kafka時，舉例。
?
?
?
?
轉(zhuǎn)載于:https://www.cnblogs.com/jareny/p/10799752.html
總結(jié)
以上是生活随笔為你收集整理的Spark-Streaming基础的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 什么是类加载器？类加载器有哪些
- 下一篇： RO38 –比较RemObjects S