小白学习spark(一些算子和算法)

一时失言乱红尘 2024-04-18 22:08 78阅读 0赞

我之前是做java开发的 , 但是进入到一个新公司 , 公司里面是大数据开发 , 都是一群大数据的大佬们 , 只有我一个java渣渣 , 所以 ,为了和同事们一起工作 , 我又开始学习spark和scala , 以下是我这几天的学习的算子和算法 (可能比较乱 , 轻喷 , 我是渣渣) , 有需要的就看看 , 莫喷我 , 只是简单记录了一下 , 代码都是自己写 , 但是 , 没有粘贴上来

1.  map （功能） 返回通过函数func传递源的每个元素形成的新分布式数据集。
2.  filter （功能） 返回通过选择func返回true 的源元素形成的新数据集。
3.  flatMap（func） 与map类似，但每个输入项可以映射到0个或更多输出项（因此func应该返回Seq而不是单个项）。
4.  mapPartitions（func） 与map类似，但在RDD的每个分区（块）上单独运行，因此当在类型T的RDD上运行时，func必须是Iterator => Iterator 类型。
5.  mapPartitionsWithIndex（func） 与mapPartitions类似，但也为func提供了表示分区索引的整数值，因此当在类型T的RDD上运行时，func必须是类型（Int，Iterator ）=> Iterator 。
6.  groupByKey（\[ numPartitions \]） 在（K，V）对的数据集上调用时，返回（K，Iterable ）对的数据集。  
    注意：如果要进行分组以便对每个密钥执行聚合（例如总和或平均值），则使用reduceByKey或aggregateByKey将产生更好的性能。  
    注意：默认情况下，输出中的并行级别取决于父RDD的分区数。您可以传递可选numPartitions参数来设置不同数量的任务。  
    groupByKey()的功能是，对具有相同键的值进行分组。比如，对四个键值对(“spark”,1)、(“spark”,2)、(“hadoop”,3)和(“hadoop”,5)，采用groupByKey()后得到的结果是：(“spark”,(1,2))和(“hadoop”,(3,5))。  
    我们对上面第二种方式创建得到的pairRDD进行groupByKey()操作，
7.  reduceByKey（func，\[ numPartitions \]） 当调用（K，V）对的数据集时，返回（K，V）对的数据集，其中使用给定的reduce函数func聚合每个键的值，该函数必须是类型（V，V）=> V.同样groupByKey，reduce任务的数量可通过可选的第二个参数进行配置。
8.  aggregateByKey（zeroValue）（seqOp，combOp，\[ numPartitions \]） 当调用（K，V）对的数据集时，返回（K，U）对的数据集，其中使用给定的组合函数和中性“零”值聚合每个键的值。  
    允许与输入值类型不同的聚合值类型，同时避免不必要的分配。同样groupByKey，reduce任务的数量可通过可选的第二个参数进行配置。
9.  getOrElse用于当集合或者option中有可能存在空值或不存在要查找的值的情况， myMap.getOrElse(“myKey”, “no such key”) myMap中没有"myKey" 时 返回"no such key"
10. scala中iterator只能执行一次迭代，如果需要多次执行同一个迭代体，建议调用iterator.toList等方法，将迭代体转化为集合，再执行上述的验证例子就会正常。
11. iterator.min和iterator.max同样是通过迭代获得，所以对于同一个iterator的min和max只能获取一个。
12. mkString(seq:String)方法是将原字符串使用特定的字符串seq分割。
13. mkString(statrt:String,seq:String,end:String)方法是将原字符串使用特定的字符串seq分割的同时，在原字符串之前添加字符串start，在其后添加字符串end。
14. foreachRDD、foreachPartition和foreach的不同之处主要在于它们的作用范围不同，  
    foreachRDD : 作用于DStream中每一个时间间隔的RDD，  
    foreachPartition : 作用于每一个时间间隔的RDD中的每一个partition，  
    foreach : 作用于每一个时间间隔的RDD中的每一个元素。
15. join：相当于mysql的INNER JOIN，当join左右两边的数据集都存在时才返回
16. leftOuterJoin：相当于mysql的LEFT JOIN，leftOuterJoin返回数据集左边的全部数据和数据集左边与右边有交集的数据
17. rightOuterJoin：相当于mysql的RIGHT JOIN，rightOuterJoin返回数据集右边的全部数据和数据集右边与左边有交集的数据
18. fullOuterJoin：返回左右数据集的全部数据，左右有一边不存在的数据以None填充
19. VertexId : 是Long的别名
20. distinct : 对RDD中的元素进行去重操作。
21. zipWithIndex或者zip : 来自动地创建一个计数器
22. Seq是列表，适合存有序重复数据，进行快速插入/删除元素等场景
23. Set是集合，适合存无序非重复数据，进行快速查找海量元素的等场景
24. 将string类型强转为数组 doc.get(“MessageOptions”).asInstanceOf\[Array\[String\]\].length
25. Spark提供了两种创建RDD的方式：读取外部数据集，以及在驱动器程序中对一个集合进行并行化。
26. 在驱动器程序中对一个集合进行并行化的方式有两种：parallelize()和makeRDD()。
27. sortByKey()的功能是返回一个根据键排序的RDD。
28. mapValues(func): 我们经常会遇到一种情形，我们只想对键值对RDD的value部分进行处理，而不是同时对key和value进行处理。对于这种情形，Spark提供了mapValues(func)，  
    它的功能是，对键值对RDD中的每个value都应用一个函数，但是，key不会发生变化。比如，对四个键值对(“spark”,1)、(“spark”,2)、(“hadoop”,3)  
    和(“hadoop”,5)构成的pairRDD，如果执行pairRDD.mapValues(x => x+1)，就会得到一个新的键值对RDD，它包含下面四个键值对(“spark”,2)、(“spark”,3)、(“hadoop”,4)和(“hadoop”,6)。
29. join(连接)操作是键值对常用的操作。“连接”(join)这个概念来自于关系数据库领域，因此，join的类型也和关系数据库中的join一样，  
    包括内连接(join)、左外连接(leftOuterJoin)、右外连接(rightOuterJoin)等。最常用的情形是内连接，所以，join就表示内连接。  
    对于内连接，对于给定的两个输入数据集(K,V1)和(K,V2)，只有在两个数据集中都存在的key才会被输出，最终得到一个(K,(V1,V2))类型的数据集。
30. 这里介绍如何把RDD保存成文本文件，后面还会介绍其他格式的保存。  
    方式一  
    val peopleDF = spark.read.format(“json”).load(“file:///usr/local/spark/examples/src/main/resources/people.json”)  
    peopleDF.select(“name”, “age”).write.format(“csv”).save(“file:///usr/local/spark/mycode/newpeople.csv”)  
    方式二  
    val peopleDF = spark.read.format(“json”).load(“file:///usr/local/spark/examples/src/main/resources/people.json”)  
    df.rdd.saveAsTextFile(“file:///usr/local/spark/mycode/newpeople.txt”)
31. rdd.top(2) rdd会自动排序,从大到小取最大两个  
    rdd.take(2) 取最小的两个  
    rdd.takeOrdered(3) 先按照升序排序 , 然后取前三个(最小的三个)
32. aggregate : 先局部聚合再整体聚合  
    aggregateByKey : 先局部进行分组 , 然后根据key进行操作 (先局部操作,在整体操作(根据key相同的 , 将value加在一起))  
    combineByKey: 跟reduceByKey运行出来的结果是一样的, 只不过, 操作比较麻烦, 比较难理解  
    combineByKey 这个方法和reduceByKey的运行结果是一样的, 都是根据key去统计value的值  
    sc.textFile(“hdfs://node-1.itcast.cn:90000/wc”).flatMap(*.split(" ")).map((*,1)).combineByKey(x=>x,(m:Int,n:Int)=>(m+n),(a:Int,b:Int)=>(a+b))
33. zip 的用法  
    val rdd1 = sc.parallelize(List(“salmon”,“rabbit”,“turkey”,“wolf”,“bear”),2)  
    val rdd2 = sc.parallelize(List(1,2,3,4,5),2)  
    val rdd3 = rdd1.zip(rdd2)//将rdd2的数量,添加到rdd1上,作为其value
34. repatition : 重新分区 (可以让分区重新指定) rdd.repatition(2)重新分成两个分区
35. 集合转换操作，以\{1,2,3\}\{3,4,5\}为例，rdd代表已生成的RDD实例  
    函数名 目的 示例 结果  
    union(rdd) 合并两个RDD所有元素（不去重） rdd1.union(rdd2) \{1,2,3,3,4,5\}  
    intersection(rdd) 求两个RDD的交集 rdd1.intersection(rdd2) \{3\}  
    substract(rdd) 移除在RDD2中存在的RDD1元素 rdd1.substract(rdd2) \{1,2\}  
    cartesian(rdd) 求两个RDD的笛卡尔积 rdd1.cartesian(rdd2) \{(1,3),(1,4),(1,5)…(3,5)\}
36. collectAsMap  
    val rdd = sc.parallelize(List((“a”,1),(“b”,2))  
    rdd.collectAsMap //map(b -> 2 , a -> 1)
37. countByKey //Map(a -> 1 , b -> 2 , c -> 4) 统计a出现1次 , b两次 ,c出现三次
38. countByValue // 是将整个作为一个value //Map( (a,1) -> 1 , (b,2) -> 2 , (c,2) -> 4) (a,1)出现一次 , (b,2)出现两次 (c,2)出现4次
39. flatMapValue //将value分开 , 和key重新拼接在一起  
    val rdd1 = sc.parallelize(List((“a”,“1 2”),(“b”,“3 4”)))  
    val rdd2 = rdd1.flatMapValues(\_.split(" ")) //结果为 array((a,1),(a,2),(b,3),(b,4))
40. keyBy 将元素的长度取出来作为key , 元素作为value  
    val rdd1 = sc.parallelize(List(“dog”,“salmon”,“rat”,“elephant”),3)  
    val rdd2 = rdd1.keyBy(\_.length).collect.mkString(",")  
    // 结果为 (3,dog),(6,salmon),(3,rat),(8,elephant)
41. keys 和 values  
    val rdd1 = sc.parallelize(List(“dog”,“tiger”,“lion”,“cat”))  
    val rdd2 = rdd1.map(x=>(x.length,x))  
    val rdd3 = rdd2.keys.collect.mkString(",") //3,5,4,3  
    val rdd4 = rdd2.values.collect.mkString(",") //dog,tiger,lion,cat
42. wordCount产生几个RDD : (产生5个RDD)  
    1. HadoopRDD (textFile)  
    2. MapPartitionsRDD (textFile) //textFile会产生两个RDD  
    3. flatMap  
    4. map  
    5. shuffledRDD (reduceByKey)
43. RDD.cache :会被重复使用的(但是)不能太大的RDD需要cache  
    RDD.checkpoint  
    对于cache，若机器发生故障，内存或者磁盘中缓存的数据丢失时，就要根据lineage(血统)进行数据恢复，想象一下，  
    如果在这之前有100个rdd，那么在要经过100次的转换，才能将数据恢复过来，这样效率非常低。

所以可以使用rdd的checkpoint机制(检查点，相当于快照)，将你认为很重要的rdd存放到一个公共的高可用的存储系统中去，  
如hdfs，下次数据丢失时，就可以从前面ck的rdd直接进行数据恢复，而不需要根据lineage去从头一个一个的去恢复，这样极大地提高了效率。  
sc.setCheckpointDir(“hdfs://master:9000/rdd-checkpoint”) //使用hdfs做存储，如果文件目录不存在会创建一个新的  
建议先将rdd缓存一下，这样会直接对内存中的数据进行ck，即： rdd2.cache().checkpoint

1.  RDD的宽依赖和窄依赖  
    窄依赖 (1对1或者1对多) : map filter  
    宽依赖 (多对多) : groupByKey
2.  DataFrame 的DSL语法  
    RDD.toDF.select(“name”,“id”).show  
    RDD.toDF.select(col(“name”),col(“id”)+1).show //必须加上()  
    RDD.toDF.filter(col(“id”)>18).show //展现age大于18的人的信息(包括所有id,name,age) 必须加上()
3.  SQL风格语法  
    RDD.toDF.registerTempTable(“表名”) //将一个dataFormate转化为一个数据库中的表进行操作  
    sqlContext.sql(“select \* from biaomi order by age desc limit 2”).show
4.  import sqlContext.implicits.\_ 必须要导进来的隐式转换