Spark基础

一、pyspark启动

指定其他python版本

 pyspark --master local --deploy-mode client --name dataplore --conf spark.pyspark.python=/data/soft/anaconda3/envs/py37/bin/python --conf spark.pyspark.driver.python=/data/soft/anaconda3/envs/py37/bin/python --driver-memory 1024M --driver-cores 1
1

如果运行脚本--conf配置放在脚本运行之前

二、RDD

aggregate

RDD-aggregate
PySpark之aggregate函数，查看分区数量getNumPartitions()和查看每个分区的内容glom()介绍
spark常用RDD算子 - cogroup
rdd算子之cogroup

combineByKey

//combineByKey方法需要三个参数：
//第一个参数表示：将相同key的第一个数据进行结构转换，实现操作
//第二个参数：分区内的计算规则
//第三个参数：分区间的计算规则

val newRDD: RDD[(String, (Int, Int))] = rdd.combineByKey(
    v => (v, 1),
    (t: (Int, Int), v) => {
       (t._1 + v, t._2 + 1)
    },
    (t1 Int: , t2: Int) => {
       (t1._1 + t2._1, t1._2 + t2._2)   
    }
) 
1
2
3
4
5
6
7
8
9

combinebykey和aggregatebykey之间的区别
reduceByKey和groupByKey的区别

localCheckpoint

巧用localCheckpoint加速Spark上的迭代计算（break lineage）

mapPartitionsWithIndex

Spark(RDD)转换操作—mapPartitionsWithIndex函数

map mapValues

PySpark map mapValues用法

saveAsHadoopFile

Spark(Streaming)写入数据到hdfs__saveAsHadoopFile以及自定义MultipleOutputFormat

treeAggregate

spark-aggregate与treeAggregate的理解

reduce

spark算子reduce,fold,aggregate的区别
groupByKey、reduceByKey、aggregateByKey区别

accumulator

Pyspark累加器(accumulator)陷阱
pyspark中使用累加器Accumulator统计指标

共享变量
Spark提供了两种共享变量：Broadcast Variable（广播变量）和Accumulator（累加变量）
BroadcastVariable会将使用到的变量，仅仅为每个节点拷贝一份，更大的用处是优化性能，减少网络传输以及内存消耗。广播变量是只读的。

val factor = 3
val broadcastVars = sc.broadcast(factor);
val numberList = Array(1,2,3,4,5)
val number = sc.parallelize(numberList).map( num => num * broadcastVars.value)  //广播变量读值broadcastVars.value
1
2
3
4

Accumulator则可以让多个task共同操作一份变量，主要可以进行累加操作。task只能对Accumulator进行累加操作，不能读取它的值。只有Driver程序可以读取Accumulator的值。

val numberList = Array(1,2,3,4,5)
val numberRDD = sc.parallelize(numberList,1)
val sum = sc.accumulator(0)
numberRDD.foreach{m => sum += m}
1
2
3
4