Spark-shell&Scala（三）map与flatMap

2023年12月26日 288次阅读 来源: sf705

刚开始看Spark API 或者Scala编程的时候，发现函数式编程看的不太明白。又不想系统的看看Scala的书，就找找网上资料了，顺便做做笔记。

map

map操作，按照Spark里面的说就是，将一个RDD中的每一个元素都执行一个指定的函数，产生一个新的RDD，两个RDD中的元素一 一对应，除此之外，生成新的RDD与原来的RDD分区个数一样。

  • map(x => x2)：x => x2 是一个函数，x是传入参数即RDD的每个元素，x*2是返回值
  • map(x => (x,1)): 该函数将RDD的每个元素（例如：a,b,c…）变为:(a,1),(b,1),(c,1)…键值对的形式
  • map(x => x.split(“\s+”)): 该函数以行为分割，将每行变为一个array。
scala> val test1 = sc.parallelize(1 to 9) //默认分区数来创建一个RDD
test1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> test1.collect //查看一下test1的内容
res2: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)

scala> val test2 = test1.map(x=>(x*2))// Int类型的每个元素乘以2
test3: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[2] at map at <console>:26

scala> test2.collect
res8: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)

scala> val test3 = test1.map(x => (x,"one"))//每个元素后加上一个字符串“one”,变成键值对。
test4: org.apache.spark.rdd.RDD[(Int, String)] = MapPartitionsRDD[3] at map at <console>:26

scala> test3.collect
res10: Array[(Int, String)] = Array((1,one), (2,one), (3,one), (4,one), (5,one), (6,one), (7,one), (8,one), (9,one))

//查看HDFS 中的数据
Spark@master:~/hadoop-2.7.0/sbin$ hadoop fs -text /test/hello.txt
hello world
nihao a 
ni zai gan ma
hello tom
hello jeey
ni zai na

scala> val test4 = sc.textFile("/test/hello.txt")//从HDFS中读取数据
test4: org.apache.spark.rdd.RDD[String] = /test/hello.txt MapPartitionsRDD[5] at textFile at <console>:24

scala> test4.map(line => line.split("\\s+")).collect
res12: Array[Array[String]] = Array(Array(hello, world), Array(nihao, a), Array(ni, zai, gan, ma), Array(hello, tom), Array(hello, jeey), Array(ni, zai, na), Array(""))

flatMap

flatMap与map有些类似，却别是map将原RDD中的每个元素处理后只生成一个新的元素，而flatMap却可以生成多个元素，还是用上面的数据，做一下对比。

scala> test4.flatMap(line => line.split("\\s+")).collect
res18: Array[String] = Array(hello, world, nihao, a, ni, zai, gan, ma, hello, tom, hello, jeey, ni, zai, na, "")

distinct

distinct去掉重复的元素

scala> test4.flatMap(line => line.split("\\s+")).distinct.collect
res24: Array[String] = Array(ma, a, tom, zai, "", jeey, hello, gan, ni, na, nihao, world)

mapPartitons

map输入是RDD中的每一个元素，而mapPartitons输入是RDD中的每个分区的迭代器，mapPartitions比map高效。

其他

  • union返回两个RDD的合并，返回元素不去重。
    rdd1.unior(rdd2)

参考

图解Spark核心技术与案例实战

    原文作者：sf705
    原文地址: https://www.jianshu.com/p/fd167ac9c87c
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞