数据读取与保存
Text文件
- 基本语法
- 数据读取：textFile(String)
- 数据保存：saveAsTextFile(String)
Sequence文件
-
SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。在SparkContext中，可以调用sequenceFile[keyClass, valueClass] (path)。
-
语法
-
保存数据为SequenceFile
dataRDD.saveAsSequenceFile("output")
-
读取SequenceFile文件
sc.sequenceFile[Int,Int] ("output").collect().foreach(println)
-
Object对象文件
- 对象文件是将对象序列化后保存的文件，采用Java的序列化机制。可以通过objectFile[k,v] (path)函数接收一个路径，读取对象文件，返回对应的RDD，也可以通过调用saveAsObjectFile()实现对对象文件的输出。因为是序列化所以要指定类型。
- 语法：
- 保存数据：dataRDD.saveAsObjectFile("output")
- 读取数据：sc.objectFile[(Int)] ("output").collect().foreach(println)
文件系统类数据读取与保存
- Spark的整个生态系统与Hadoop是完全兼容的，所以对于Hadoop所支持的文件类型或者数据库类型，Spark也同样支持。
累加器
- 累加器：分布式共享只写变量。（Executor和Executor之间不能读数据）
- 累加器用来把Executor端变量信息聚合到Driver端。在Driver程序中定义的变量，在Executor端的每个task都会得到这个变量的一份新的副本，每个task更新这些副本的值后，传回Driver端进行merge。
- 原因: 累加器可以在一定程度上减少shuffle操作
- 场景: 用于聚合结果不是太大的场景，不然可能导致Driver内存溢出
- 原理: 首先在每个分区中聚合,然后将每个分区的聚合结果发给Driver汇总
系统累加器
累加器使用
-
累加器定义（SparkContext.accumulator(initialValue)方法）
val sum: LongAccumulator = sc.longAccumulator("sum")
-
累加器添加数据（累加器.add方法）
sum.add(count)
-
累加器获取数据（累加器.value）
sum.value
-
注意：Executor端的任务不能读取累加器的值（例如：在Executor端调用sum.value，获取的值不是累加器最终的值）。从这些任务的角度来看，累加器是一个只写变量。
-
累加器放在行动算子中：放在foreach()这样的行动操作中，这样无论计算失败还是计算重复累加器都相对可靠。转化操作中累加器可能会发生不止一次更新。
自定义累加器
广播变量
- 原因: spark算子里面的代码是在executor的task执行的,spark算子外面的代码是在driver执行
- 场景:
- spark算子中使用了driver的数据的时候,并且该数据的大小还有点大,此时可以将数据广播出去减少数据
- 大表 join 小表的时候,将小表广播出去减少shuffle操作
- 使用:
- 广播数据: val bc = sc.broadcast(数据)【大表join小表的时候需要使用collect收集小表数据然后才能广播。即不能广播RDD，RDD中没有数据，只有计算逻辑】
- task使用广播数据: bc.value
场景一代码演示
//spark算子中使用了driver的数据的时候,并且该数据的大小还有点大,此时可以将数据广播出去减少数据
def main1(args: Array[String]): Unit = {
import org.apache.spark.{SparkConf, SparkContext}
val sc = new SparkContext( new SparkConf().setMaster("local[4]").setAppName("test") )
val rdd = sc.parallelize(List("jd","pdd","tm","atguigu"))
//
val map = Map[String,String]("jd"->"www.jd.com","pdd"->"www.pdd.com","tm"->"www.tm.com","atguigu"->"www.atguigu.com")
//广播数据
val bc = sc.broadcast(map)
val rdd2 = rdd.map(x=> {
val m = bc.value
m.getOrElse(x,"")
//map.getOrElse(x,"")
})
println(rdd2.collect().toList)
Thread.sleep(1000000)
}
场景二代码演示
//大表 join 小表的时候,将小表广播出去减少shuffle操作
def main(args: Array[String]): Unit = {
import org.apache.spark.{SparkConf, SparkContext}
val sc = new SparkContext( new SparkConf().setMaster("local[4]").setAppName("test") )
//获取没有农贸市场的省份
//读取数据
//全国省份信息
val allprovincerdd = sc.textFile("datas/allprovince.txt")
//全国部分省份菜市场信息
val productRdd = sc.textFile("datas/product.txt")
/*大概长这样，数据量挺大
生菜 2.00 2018/1/1 山西汾阳市晋阳农副产品批发市场 山西 汾阳
芹菜 2.40 2018/1/1 山西汾阳市晋阳农副产品批发市场 山西 汾阳
菜花 3.80 2018/1/1 北京朝阳区大洋路综合市场 北京 朝阳
生姜 10.00 2018/1/1 北京朝阳区大洋路综合市场 北京 朝阳
*/
//过滤，因为有些数据不太干净，缺失了部分，在此不考虑这些缺失数据
val productFilterRdd = productRdd.filter(line=> line.split("\t").length==6)
//列裁剪，获取需要的省份数据
val provinceRdd = productFilterRdd.map(line=>{
val arr = line.split("\t")
arr(4)
})
//对切分后的数据进行去重
val disProvinceRdd = provinceRdd.distinct()
//将处理后的数据转换成KV的形式
val provinceKVRDD = disProvinceRdd.map(x=>(x,""))
//将所有省份的数据转换成KV的形式
val allProvinceKVRDD = allprovincerdd.map(x=>(x,""))
//广播经过处理得到的全国所有菜市场省份数据
val provinceList = provinceKVRDD.collect
val bc = sc.broadcast( provinceList )
//val allRdd = allProvinceKVRDD.leftOuterJoin(provinceKVRDD)
//val resRdd = allRdd.filter{
// case (province,( leftvalue,rightValue )) => rightValue.isEmpty
//}
val resRdd = allProvinceKVRDD.filter{
case (province,_) =>
! bc.value.map(_._1).contains( province )
}
resRdd.foreach(println)
Thread.sleep(10000000)
}