spark 的一些常用函数 filter，map，flatMap，lookup ，reduce，groupByKey

定义不带参数也不带返回值的函数（def :定义函数的关键字  printz:方法名称）

scala> def printz = print("scala hello")

定义带参数也带返回值的函数（这种函数在定义时也可以不带返回值的类型，scala会自动推算出。建议还是带上）

scala> def minNum(x:Int,y:Int):Int = if(x>y) x else y    //:Int 是该函数的返回值类型
minNum: (x: Int, y: Int)Int

调用函数

scala> minNum(2,3)
res10: Int = 3

创建一个map

scala> val colors = Map("red" -> "#FF0000", "azure" -> "#F0FFFF")
colors: scala.collection.immutable.Map[String,String] = Map(red -> #FF0000, azur
e -> #F0FFFF)

将map中的每一个元素中的第一个元组与第二个元组交换位置

‘_‘：表示map的元组，

x._2：map中每一个元素的第二个元组（#FF0000,"#F0FFFF"）

scala> colors.map(x=>(x._2,x._1))
res0: scala.collection.immutable.Map[String,String] = Map(#FF0000 -> red, #F0FFF
F -> azure)

union(合并rdd)

scala> var rdd1=sc.parallelize(List(1,2,3,4))//创建rdd1
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[53] at parallelize at <console>:24
scala> var rdd2=sc.parallelize(List(5,6,7,8))//创建rdd2
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[54] at parallelize at <console>:24
scala> var unionRes=rdd1 union rdd2  //合并rdd
unionRes: org.apache.spark.rdd.RDD[Int] = UnionRDD[55] at union at <console>:28
scala> unionRes.collect　　　　　  　//查合并后的rdd
res19: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8)

join 类似于sql中的内连，左外链接，右外链接

创建rdd

scala> var rdd1=sc.parallelize(List((1,"spark"),(2,"hadoop"),(3,"scala"),(4,"java"))) //创建rdd1
rdd1: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[58] at parallelize at <console>:24
scala> var rdd2=sc.parallelize(List((1,"30k"),(2,"28k"),(3,"5k"),(5,"10k")))　　　　　　//创建rddd2
rdd2: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[59] at parallelize at <console>:24
scala> var joinRes=rdd1 join rdd2  　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　//join rdd
joinRes: org.apache.spark.rdd.RDD[(Int, (String, String))] = MapPartitionsRDD[62] at join at <console>:28
scala> joinRes.collect　　　//查看join后的结果，可以看出join会把两个rdd有相同key的元素的值进行合并
res20: Array[(Int, (String, String))] = Array((1,(spark,30k)), (3,(scala,5k)), (2,(hadoop,28k)))

因为第4和第5没有匹配项，所以不显示

leftOuterJoin（左外链接）

//这个是以左边的rdd为主，右边为辅。右边与左边没有匹配项，则会显示左边的值，所以这里key为4的也会显示
scala> var leftJoinRes=rdd1.leftOuterJoin(rdd2)
leftJoinRes: org.apache.spark.rdd.RDD[(Int, (String, Option[String]))] = MapPartitionsRDD[65] at leftOuterJoin at <console>:28
scala> leftJoinRes.collect
res21: Array[(Int, (String, Option[String]))] = Array((4,(java,None)), (1,(spark,Some(30k))), (3,(scala,Some(5k))), (2,(hadoop,Some(28k))))

rightOuterJoin(右外链接)

//这个是以右边边的rdd为主，左边边为辅。右边与左边没有匹配项，则只显示右边,所以这里key为5的也会显示
scala> var rightJoinRes=rdd1.rightOuterJoin(rdd2)
rightJoinRes: org.apache.spark.rdd.RDD[(Int, (Option[String], String))] = MapPartitionsRDD[68] at rightOuterJoin at <console>:28
scala> rightJoinRes.collect
res22: Array[(Int, (Option[String], String))] = Array((1,(Some(spark),30k)), (3,(Some(scala),5k)), (5,(None,10k)), (2,(Some(hadoop),28k)))

groupByKey

//(4,1), (7,4), (6,3), (2,2), (3,3), (1,3)

scala> f1.flatMap(x=>x.split("-")).map((_,1)).groupByKey.collect
res28: Array[(String, Iterable[Int])] = Array((4,CompactBuffer(1)), (7,CompactBuffer(1, 1, 1, 1)), (6,CompactBuffer(1, 1, 1)), (2,CompactBuffer(1, 1)), (3,CompactBuffer(1, 1, 1)), (1,CompactBuffer(1, 1, 1)))

reduce

scala> var rdd1=sc.parallelize(List(1,2,3,4,5))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[80] at parallelize at <console>:24
scala> rdd1.reduce(_+_)
res32: Int = 15

reduce(_+_):每个元组相加 1+2+3+4+5=15

他是每次相加两个元组然后产生新的rdd与下一位元组相加

1+2=3

3+3=6

6+4=10

10+5=15

lookup 

scala> var rdd1=sc.parallelize(List((1,"a"),(2,"b"),(3,"c")))
rdd1: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[81] at parallelize at <console>:24
scala> rdd1.lookup(1)
res34: Seq[String] = WrappedArray(a)

lookup函数对<key,value>型的rdd操作，返回指定key对应的元素形成的seq，这个函数的优点

如果这个rdd包含分区器，那么只扫描对应key所在的分区，然后返回对应key的元素形成的seq；如果这个rdd没有分区器，则对这个rdd进行全盘扫描，然后返回对应key的元素形成的seq

map和flatMap

map函数是以一行数据为一个元素

Array[Student] =
Array(Student(zhangxs,24,chenxy), Student(wangYr,21,teacher), Student(wangx,26,teacher))

map是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD。任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

scala> val a = sc.parallelize(1 to 9, 3)
scala> val b = a.map(x => x*2)
scala> a.collect
res10: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
scala> b.collect
res11: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)

上述例子中把原RDD中每个元素都乘以2来产生一个新的RDD。

flatMap是一个文件的数据为一个元素

res9: Array[String] = Array(zhangxs, 24, chenxy, wangYr, 21, teacher, wangx, 26, teacher)

mapPartitions

mapPartitions是map的一个变种。map的输入函数是应用于RDD中每个元素，而mapPartitions的输入函数是应用于每个分区，也就是把每个分区中的内容作为整体来处理的。 
它的函数定义为：




    

  1. def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
    2. 






f即为输入函数，它处理每个分区里面的内容。每个分区中的内容将以Iterator[T]传递给输入函数f，f的输出结果是Iterator[U]。最终的RDD由所有分区经过输入函数处理后的结果合并起来的。


比如


scala> val a = sc.parallelize(1 to 9, 3)
scala> def myfunc[T](iter: Iterator[T]) : Iterator[(T, T)] = {
var res = List[(T, T)]()
var pre = iter.next while (iter.hasNext) {
val cur = iter.next;
res .::= (pre, cur) pre = cur;
}
res.iterator
}
scala> a.mapPartitions(myfunc).collect
res0: Array[(Int, Int)] = Array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8))


上述例子中的函数myfunc是把分区中一个元素和它的下一个元素组成一个Tuple。因为分区中最后一个元素没有下一个元素了，所以(3,4)和(6,7)不在结果中。 
mapPartitions还有些变种，比如mapPartitionsWithContext，它能把处理过程中的一些状态信息传递给用户指定的输入函数。还有mapPartitionsWithIndex，它能把分区的index传递给用户指定的输入函数。



mapValues


mapValues顾名思义就是输入函数应用于RDD中Kev-Value的Value，原RDD中的Key保持不变，与新的Value一起组成新的RDD中的元素。因此，该函数只适用于元素为KV对的RDD。


scala> val a = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", " eagle"), 2)
scala> val b = a.map(x => (x.length, x))　　//key：元组的长度 value：元组的值
scala> b.mapValues("x" + _ + "x").collect //“_”就是每个元组的value，在每个value的前后加"x"
res5: Array[(Int, String)] = Array((3,xdogx), (5,xtigerx), (4,xlionx),(3,xcatx), (7,xpantherx), (5,xeaglex))



mapWith


mapWith是map的另外一个变种，map只需要一个输入函数，而mapWith有两个输入函数。它的定义如下：


def mapWith[A: ClassTag, U: ](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => U): RDD[U]



    

  • 第一个函数constructA是把RDD的partition index（index从0开始）作为输入，输出为新类型A；
    • 

  • 第二个函数f是把二元组(T, A)作为输入（其中T为原RDD中的元素，A为第一个函数的输出），输出类型为U。
    • 



比如把partition index 乘以10，然后加上2作为新的RDD的元素。


val x = sc.parallelize(List(1,2,3,4,5,6,7,8,9,10), 3)
x.mapWith(a => a * 10)((a, b) => (b + 2)).collect
res4: Array[Int] = Array(2, 2, 2, 12, 12, 12, 22, 22, 22, 22)



				


				
时间： 2024-10-11 23:07:21 

		


		


	

	
	

		


		
spark 的一些常用函数 filter，map，flatMap，lookup ，reduce，groupByKey的相关文章


								

		

			

                Python 内置函数&amp;filter()&amp;map()&amp;reduce()&amp;sorted()
			

		

		

									

				常用内置函数 Python 2.x 返回列表,Python 3.x 返回迭代器 在进行筛选或映射时,输出的结果是一个数组,需要list帮助. 如:print(list(map(lambda x:x+1, [1,2,3]))) 一.filter() --过滤.筛选 刚接触filter时 ,运行总是出现<filter object at 0x000001B68F052828> 得不到想要的数据,后来发现是因为filter的结果是一个数组, 需要 list 帮助,后来将print(f) 改为 pri			

		

	

				

		

			

                Python内置函数filter, map, reduce
			

		

		

									

				filter.map.reduce,都是对一个集合进行处理,filter很容易理解用于过滤,map用于映射,reduce用于归并. 是Python列表方法的三架马车. 1. filter函数的功能相当于过滤器. filter函数的定义: filter(function or None, sequence) -> list, tuple, or string function是一个谓词函数,接受一个参数,返回布尔值True或False. filter函数会对序列参数sequence中的每个元素调用			

		

	

				

		

			

                python|高级函数|filter|map|reduce|sorted
			

		

		

									

				filter(func, iterable) 循环调用输入的函数 过滤传入的参数,函数的结果返回的是true那就保存,返回false就不要,且返回的也是迭代器 备注: 迭代器用完一个就扔掉一个,直到全部用完: 可以用list()转化为列表:不转化则返回的为迭代器对象,可以用for循环直接逐个调用 # utils/core.py convert_legacy_filters_into_adhoc()for filt in filter(lambda x: x is not None, fd[fil			

		

	

				

		

			

                Python 函数 filter()  map()  reduce()
			

		

		

									

				1.filter(bool_func,seq) filter()是'筛选函数',也接收一个函数和一个序列,filter()把传人的函数依次作用于序列的每个元素,然后根据返回值是True还是false决定保留还是丢弃该元素 例子: def fr(x): return x%2==1 print 'filter1:',filter(fr,range(1,51))#筛选出100以内的所有奇数 print 'filter2:',filter(fr,[1,2,3,4]) 输出: filter1: [1, 3			

		

	

				

		

			

                python关于list的三个内置函数filter(), map(), reduce()
			

		

		

									

				''' Python --version :Python 2.7.11 Quote : https://docs.python.org/2/tutorial/datastructures.html#more-on-lists Add by camel97 2017-04 ''' 1.filter() #filter(function, sequence) returns a sequence consisting of those items from the sequence for whic			

		

	

				

		

			

                python内置函数filter(),map(),reduce()笔记
			

		

		

									

				'''python reduce()函数:reduce()函数会对参数序列中元素进行积累. 函数将一个数据集合(链表,元组等)中的所有数据进行下列操作:用传给reduce中的函数 function(有两个参数)先对集合中的第 1.2 个元素进行操作,得到的结果再与第三个数据用 function 函数运算,最后得到一个结果. 语法:ruduce()reduce(function,iterable,initializer)参数:function-函数,有两个参数iterable--可迭代对象init			

		

	

				

		

			

                ES6高阶函数:过滤器&gt;&gt;filter, map,reduce
			

		

		

									

				1 // 高阶函数 filter/map/reduce 2 3 // filter中的回调函数有一个要求:必须返回一个boolean值, 4 // 当返回true时,函数内部会自动将这次回调的 n 加入到新的数组中 5 // 当返回false时,函数内部会自动过滤掉这次的 n 6 const nums=[10,20,60,94,348,57,24,674,645,44,4] 7 let newNums = nums.filter(function (n){ 8 return n<30 9 })			

		

	

				

		

			

                js高阶函数filter、map、reduce
			

		

		

									

				1 // 高阶函数 filter/map/reduce 2 3 // filter中的回调函数有一个要求:必须返回一个boolean值, 4 // 当返回true时,函数内部会自动将这次回调的 n 加入到新的数组中 5 // 当返回false时,函数内部会自动过滤掉这次的 n 6 const nums=[10,20,60,94,348,57,24,674,645,44,4] 7 let newNums = nums.filter(function (n){ 8 return n<30 9 })			

		

	

				

		

			

                PYTHON-常用的类型转换函数和序列常用函数有哪些？
			

		

		

									

				一.类型转换函数 chr(i)chr()函数返回ASCII码对应的字符串.print chr(65) Aprint chr(66) Bprint chr(65)+chr(66) AB ord(x)ord()函数返回一个字符串参数的ASCII码或Unicode值.ord("a") 97ord(u"a") 97 hex(x)hex()函数可把整数转换成十六进制数.hex(16) '0x10'hex(123) '0x7b' oct(x)oct()函数可把给出的整数转换成八