val  conf = new   SparkConf (). setAppName ( "Test" ). setMaster ( "local" )
       val  sc = new   SparkContext (conf)

     // 通过并行化生成 rdd
     val  rdd = sc. parallelize (List( 5 , 6 , 4 , 7 , 3 , 8 , 2 , 9 , 10 ))

     //map: 对 rdd 里面每一个元乘以 2 然后排序
     val  rdd2: RDD[Int] = rdd. map (_ * 2 )
     //collect 以数组的形式返回数据集的所有元素 ( 是 Action 算子 )
     println (rdd2. collect (). toBuffer )

     //filter: 该 RDD 由经过 func 函数计算后返回值为 true 的输入元素组成
     val  rdd3: RDD[Int] = rdd2. filter (_ > 10 )
     println (rdd3. collect (). toBuffer )

     val  rdd4 = sc. parallelize (Array( "a b c" , "b c d" ))
     //flatMap: 将 rdd4 中的元素进行切分后压平
     val  rdd5: RDD[String] = rdd4. flatMap (_. split ( " " ))
     println (rdd5. collect (). toBuffer )
     // 假如 : List(List(" a,b" ,"b c"),List("e c"," i o"))
     // 压平 flatMap(_.flatMap(_.split(" ")))
    
     //sample 随机抽样
     //withReplacement 表示是抽出的数据是否放回， true 为有放回的抽样， false 为无放回的抽样
     //fraction 抽样比例例如 30% 即 0.3 但是这个值是一个浮动的值不准确
     //seed 用于指定随机数生成器种子 默认参数不传
     val  rdd5_ 1  = sc. parallelize ( 1  to 10 )
     val  sample = rdd. sample ( false , 0.5 )
     println (sample. collect (). toBuffer )

     //union: 求并集
     val  rdd6 = sc. parallelize (List( 5 , 6 , 7 , 8 ))
     val  rdd7 = sc. parallelize (List( 1 , 2 , 5 , 6 ))
     val  rdd8 = rdd6 union rdd7
     println (rdd8. collect . toBuffer )

     //intersection: 求交集
     val  rdd9 = rdd6 intersection rdd7
     println (rdd9. collect . toBuffer )

     //distinct: 去重出重复
     println (rdd8. distinct . collect . toBuffer )

     //join 相同的 key 会被合并
     val  rdd10_ 1  = sc. parallelize (List(( "tom" , 1 ),( "jerry"  , 3 ),( "kitty" , 2 )))
     val  rdd10_ 2  = sc. parallelize (List(( "jerry"  , 2 ),( "tom" , 2 ),( "dog" , 10 )))
     val  rdd10_ 3  = rdd10_ 1  join rdd10_ 2
     println (rdd10_ 3. collect (). toBuffer )
    
     // 左连接和右连接
     // 除基准值外是 Option 类型 , 因为可能存在空值所以使用 Option
     val  rdd10_ 4  = rdd10_ 1  leftOuterJoin rdd10_ 2   // 以左边为基准没有是 null
     val  rdd10_ 5  = rdd10_ 1  rightOuterJoin rdd10_ 2   // 以右边为基准没有是 null
     println (rdd10_ 4. collect (). toList )
     println (rdd10_ 5. collect (). toBuffer )

     val  rdd11_ 1  = sc. parallelize (List(( "tom" , 1 ),( "jerry"  , 3 ),( "kitty" , 2 )))
     val  rdd11_ 2  = sc. parallelize (List(( "jerry"  , 2 ),( "tom" , 2 ),( "dog" , 10 )))
     // 笛卡尔积
     val  rdd11_ 3  = rdd11_ 1  cartesian rdd11_ 2
     println (rdd11_ 3. collect. toBuffer )
  
    // 根据传入的参数进行分组
     val  rdd11_ 5 _ 1  = rdd11_ 4. groupBy (_._ 1 )
     println (rdd11_ 5 _ 1. collect (). toList )

     // 按照相同 key 进行分组 , 并且可以制定分区
     val  rdd11_ 5 _ 2  = rdd11_ 4. groupByKey
     println (rdd11_ 5 _ 2. collect (). toList )

     // 根据相同 key 进行分组 [ 分组的话需要二元组 ]
     //cogroup 和 groupBykey 的区别
     //cogroup 不需要对数据先进行合并就以进行分组 得到的结果是 同一个 key 和不同数据集中的数据集合
     //groupByKey 是需要先进行合并然后在根据相同 key 进行分组
     val  rdd11_ 6 : RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd11_ 1  cogroup rdd11_ 2
     println (rdd11_ 6 )

val  conf = new   SparkConf (). setAppName ( "Test" ). setMaster ( "local[*]" )
     val  sc = new   SparkContext (conf)
     /* Action 算子 */
     // 集合函数
     val  rdd1 = sc. parallelize (List( 2 , 1 , 3 , 6 , 5 ), 2 )
     val  rdd1_ 1  = rdd1. reduce (_+_)
     println (rdd1_ 1 )
     // 以数组的形式返回数据集的所有元素
     println (rdd1. collect (). toBuffer )
     // 返回 RDD 的元素个数
     println (rdd1. count ())
     // 取出对应数量的值 默认降序 , 若输入 0 会返回一个空数组
     println (rdd1. top ( 3 ). toBuffer )
     // 顺序取出对应数量的值
     println (rdd1. take ( 3 ). toBuffer )
     // 顺序取出对应数量的值 默认生序
     println (rdd1. takeOrdered ( 3 ). toBuffer )
     // 获取第一个值 等价于 take(1)
     println (rdd1. first ())
     // 将处理过后的数据写成文件 ( 存储在 HDFS 或本地文件系统 )
     //rdd1.saveAsTextFile("dir/file1")
     // 统计 key 的个数并生成 map k 是 key 名 v 是 key 的个数
     val  rdd2 = sc. parallelize (List(( "key1" , 2 ),( "key2" , 1 ),( "key3" , 3 ),( "key4" , 6 ),( "key5" , 5 )), 2 )
     val  rdd2_ 1 : collection. Map [String, Long] = rdd2. countByKey ()
     println (rdd2_ 1 )
     // 遍历数据
    rdd1. foreach (x => println (x))

     /* 其他算子 */
     // 统计 value 的个数 但是会将集合中的一个元素看做是一个 vluae
     val  value: collection. Map [(String, Int), Long] = rdd2. countByValue
     println (value)
     //filterByRange: 对 RDD 中的元素进行过滤 , 返回指定范围内的数据
     val  rdd3 = sc. parallelize (List(( "e" , 5 ),( "c" , 3 ),( "d" , 4 ),( "c" , 2 ),( "a" , 1 )))
     val  rdd3_ 1 : RDD[(String, Int)] = rdd3. filterByRange ( "c" , "e" ) // 包括开始和结束的
     println (rdd3_ 1. collect. toList )
     //flatMapValues 对参数进行扁平化操作 , 是 value 的值
     val  rdd3_ 2  = sc. parallelize (List(( "a" , "1 2" ),( "b" , "3 4" )))
     println ( rdd3_ 2.f latMapValues(_. split ( " " )). collect . toList )
     //foreachPartition 循环的是分区数据
     // foreachPartiton 一般应用于数据的持久化 , 存入数据库 , 可以进行分区的数据存储
     val  rdd4 = sc. parallelize (List( 1 , 2 , 3 , 4 , 5 , 6 , 7 , 8 , 9 ), 3 )
    rdd4. foreachPartition (x => println (x. reduce (_+_)))
     //keyBy 以传入的函数返回值作为 key ,RDD 中的元素为 value 新的元组
     val  rdd5 = sc. parallelize (List( "dog" , "cat" , "pig" , "wolf" , "bee" ), 3 )
     val  rdd5_ 1 : RDD[(Int, String)] = rdd5. keyBy (_. length )
     println (rdd5_ 1. collect. toList )
     //keys 获取所有的 key  values 获取所有的 values
     println (rdd5_ 1. keys. collect . toList )
     println (rdd5_ 1. values. collect . toList )
     //collectAsMap   将需要的二元组转换成 Map
     val  map: collection. Map [String, Int] = rdd2. collectAsMap ()
     println (map)