1 package com.dingxin.datainit 2 3 import org.apache.log4j.{Level, Logger} 4 import org.apache.spark.sql.SparkSession 5 6 /** 7 * Created by zhen on 2018/12/18. 8 */ 9 object RDDTransform { 10 def main(args: Array[String]) { 11 Logger.getLogger("org.apache.spark").setLevel(Level.WARN) // 设置日志级别 12 val spark = SparkSession.builder().appName("算子").master("local[2]").getOrCreate() 13 // 算子:maxBy 14 val sourceDate = Array("zhen@zhen01/2018-09-04_18;57;02_SOURCE","zhen@zhen02/2018-09-05_11;37;11_SOURCE", 15 "zhen@zhen03/2018-09-06_11;37;11_TEST") 16 17 val max_by_key = sourceDate 18 .map(_.split("/").last) 19 .filter(x => x.split("_").last.equals("SOURCE")) 20 .maxBy(x => x.split("_").take(2).mkString(""))// 根据输入的条件选择最大值minBy与之相似 21 println("------maxBy算子------") 22 max_by_key.foreach(print) 23 println("\n------flatten算子------") 24 // 算子:flatten 展开多层集合为一个List 25 val resultDate = Array("zhen@zhen01/2018-09-04_18;57","zhen@zhen02/2018-09-05_11;37","zhen@zhen03/2018-09-06_11;37") 26 val seq = Seq(sourceDate, resultDate) 27 seq.flatten.map(r => r + "_plus").foreach(println) 28 // 算子:cogroup 对两个内部数据结构为元组(仅有两个元素的元组)的数据进行匹配,把匹配上的value值保存到一个元组中 29 println("------cogroup算子------") 30 val sc = spark.sparkContext 31 val key = sc.parallelize(Seq("spark", "scala", "hadoop","scala", "python", "AI")) 32 val key_plus = sc.parallelize(Seq("spark", "scala", "python")) 33 val value = sc.parallelize(Seq(12, 3, 16,23, 21, 45)) 34 val value_plus = sc.parallelize(Seq(16, 13, 6)) 35 val tuple_data = key.zip(value) 36 val tuple_plus_data = key_plus.zip(value_plus) 37 tuple_data.cogroup(tuple_plus_data).collect().foreach(println) 38 39 } 40 }
结果: