背景

本文使用idea编程

spark 版本

<scala.version>2.11.8</scala.version>
<spark.version>2.2.0</spark.version>
<scala.compat.version>2.11</scala.compat.version>

备注

<ps>Spark中，只有遇到action，才会执行RDD的计算(即延迟计算)</ps>

RDD创建

创建方式

从集合中创建RDD

从外部存储创建RDD

从其他RDD创建

  def testCreate(spark: SparkSession) = {val sc: SparkContext = spark.sparkContext// 使用makeRDD从集合中创建val rdd1: RDD[(Int, Int)] = sc.makeRDD(Array((1, 2), (2, 3), (3, 4), (4, 5)))// 使用parallelize从集合中创建val rdd2: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4, 5))// 从外部文件中获取(还可以从各种数据库、HDFS等等外部资源中获取)val rdd3: RDD[String] = sc.textFile("***/RddDemo.scala")//从rdd2中获取新的rddval rdd4: RDD[Int] = rdd2.map(x => x + 2)}

RDD转换

Value类型交互

map(fun)

作用：返回一个新的rdd，每个元素根据fun函数进行映射

    // List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)val rdd1: RDD[Int] = spark.sparkContext.parallelize(1 to 10)//     可以简写成这个样子rdd1.map(_ + 2)//List(3, 4, 5, 6, 7, 8, 9, 10, 11, 12)val rdd2: RDD[Int] = rdd1.map(x => x + 2)

mapPartitions(fun)

作用：类似map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，fun的函数类型必须是Iterator[T] => Iterator[U]

PS:假设有N个元素，有M个分区，那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区

    //List(3, 4, 5, 6, 7, 8, 9, 10, 11, 12)val rdd3: RDD[Int] = rdd1.mapPartitions(x => x.map(e => e + 2))

map()和 mapPartitions()的区别

map 每次只处理一条数据

mapPartition每次处理一个分区的数据，这个分区的数据处理完后，原RDD中分区的数据才释放，容易导致OOM

当内存足够大的时候建议使用mapPartition，可以提升效率

mapPartitionsWithIndex(fun)

作用：类似于mapPartitions，但fun带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，fun的函数类型必须是(Int, Interator[T]) => Iterator[U]

PS：index不会超过分片个数，如果没有指定分片的话，系统有默认分片

    //List((0,1), (1,2), (2,3), (3,4), (3,5), (4,6), (5,7), (6,8), (7,9), (7,10))val rdd4: RDD[(Int, Int)] = rdd1.mapPartitionsWithIndex((index, x) => x.map(e => (index, e)))

flatMap(fun)

作用：扁平化映射，每个元素可以映射成0活多个元素(所以函数输出的是一个序列)

    //List(-2, -1, 0, 1, -1, 0, 1, 2, 0, 1, 2, 3, 1, 2, 3, 4, 2, 3, 4, 5, 3, 4, 5, 6, 4, 5, 6, 7, 5, 6, 7, 8, 6, 7, 8, 9, 7, 8, 9, 10)val rdd5: RDD[Int] = rdd1.flatMap(x => x - 3 to x)

glom()

作用：将每一个分区形成一个数组，形成新的RDD类型时RDD[Array[T]]

PS：如果rdd中没有指定分片个数，系统会有默认的分片个数(根据系统资源默认获取)

    //List(List(3), List(4, 5), List(7), List(1), List(6), List(2), List(9, 10), List(8))// 没有指定指定分片数量val rdd6: RDD[Array[Int]] = rdd1.glom()val r1: RDD[Int] = sc.parallelize(1 to 10, 3)//List(List(1, 2, 3), List(7, 8, 9, 10), List(4, 5, 6))// 指定分片数量val rdd61: RDD[Array[Int]] = r1.glom()

groupBy(fun)

作用：分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器

    //List((0,CompactBuffer(3, 6, 9)), (1,CompactBuffer(1, 4, 7, 10)), (2,CompactBuffer(2, 5, 8)))//每个元素对3取余分组val rdd7: RDD[(Int, Iterable[Int])] = rdd1.groupBy(x => x % 3)

filter(fun)

作用：过滤。返回一个新的RDD，该RDD由经过fun函数计算返回值为true的输入元素组成

    //List(2, 4, 6, 8, 10)//求全部偶数集合val rdd8: RDD[Int] = rdd1.filter(x => x % 2 == 0)

distinct([numTasks]))

作用：对源RDD进行去重后返回一个新的RDD。默认情况下，只有8个并行任务来操作，但是可以传入一个可选的numTasks参数改变它

PS: 指定不同的并行度得到的数据顺序可能不一致

    val distinctRdd: RDD[Int] = sc.makeRDD(Array(1, 1, 2, 2, 3, 3, 4, 4, 5))// 可以指定并行度 得到的结果顺序可能不一致// List(1, 2, 3, 4, 5)val rdd9: RDD[Int] = distinctRdd.distinct()// List(3, 4, 1, 5, 2)val rdd91: RDD[Int] = distinctRdd.distinct(3)

sortBy(func,[ascending], [numTasks])

作用：使用func先对数据进行处理，按照处理后的数据比较结果排序，默认为正序

    //排序 -倒序val sorted1: RDD[Int] = rdd1.sortBy(x => x, false, 4)println("操作后数据： " + sorted1.collect().toList)val s1: RDD[(Int, Int, Int)] = sc.makeRDD(Array((1, 2, 6), (6, 5, 3), (1, 2, 3), (9, 6, 6), (6, 7, 3)))// 按照第三个参数 第二个参数 第一个参数顺序排序val sorted2: RDD[(Int, Int, Int)] = s1.sortBy(x => (x._3, x._2, x._1), false)val sorted3: Array[(Int, Int, Int)] = s1.top(10)(Ordering[(Int, Int, Int)].on(x => (x._3, x._2, x._1)))//整数排序：1: List(10, 9, 8, 7, 6, 5, 4, 3, 2, 1)
//多参数排序2: List((9,6,6), (1,2,6), (6,7,3), (6,5,3), (1,2,3))
//多参数排序3: List((9,6,6), (1,2,6), (6,7,3), (6,5,3), (1,2,3))

coalesce(numPartitions)

作用：缩减分区数，用于大数据集过滤后，提高小数据集的执行效率

repartition(numPartitions)

作用：根据分区数，重新通过网络随机洗牌所有数据

    val r2: RDD[Int] = sc.parallelize(1 to 10, 4)val rdd10: RDD[Int] = r2.coalesce(1)println("之前分区数量： " + r2.partitions.length + " ;调整之后：" + rdd10.partitions.length)val rdd11: RDD[Int] = r2.repartition(3)println("之前分区数量： " + r2.partitions.length + " ;调整之后：" + rdd11.partitions.length)
//之前分区数量： 4 ;coalesce调整之后：1
//之前分区数量： 4 ;repartition调整之后：3

区别和联系

repartition实际上是调用的coalesce，默认是进行shuffle的

coalesce重新分区，可以选择是否进行shuffle过程。由参数shuffle: Boolean = false/true决定

sample(withReplacement, fraction, seed)

作用：以指定的随机种子随机抽样出数量为fraction的数据，withReplacement表示是抽出的数据是否放回，true为有放回的抽样，false为无放回的抽样，seed用于指定随机数生成器种子

  //放回抽样val rdd12: RDD[Int] = rdd1.sample(true, 0.5, 2)println("放回抽样：" + rdd12.collect().toList)val rdd13: RDD[Int] = rdd1.sample(false, 0.5, 2)println("不放回抽样：　" + rdd13.collect().toList)
//放回抽样：List(3, 5, 7, 9, 9)
//不放回抽样：　List(2, 3, 5, 8, 10)

pipe(command, [envVars])

作用：管道，针对每个分区，都执行一个shell脚本，返回输出的RDD。

注意：脚本需要放在Worker节点可以访问到的位置

    // 可以是编译器内的绝对路径，window环境不能执行 linux的shell脚本val rdd14: RDD[String] = rdd1.pipe("***\\pip.bat")println("管道执行：" + rdd14.collect().toList)

双Value类型交互

union(otherDataset)

作用：【并集】对源RDD和参数RDD求并集后返回一个新的RDD

subtract (otherDataset)

作用：【差集】计算差的一种函数，去除两个RDD中相同的元素，不同的RDD将保留下来

intersection(otherDataset)

作用：【交集】对源RDD和参数RDD求交集后返回一个新的RDD

zip(otherDataset)

作用：【拉链】将两个RDD组合成Key/Value形式的RDD,这里默认两个RDD的partition数量以及元素数量都相同，否则会抛出异常

cartesian(otherDataset)

作用：【笛卡尔积】尽量避免使用

 def testTransferDoubleValue(spark: SparkSession) = {val sc: SparkContext = spark.sparkContextval rdd1: RDD[Int] = sc.parallelize(1 to 4)val rdd2: RDD[Int] = sc.parallelize(2 to 5)//求并集val union: RDD[Int] = rdd1.union(rdd2)println("数据源1: " + rdd1.collect().toList)println("数据源2: " + rdd2.collect().toList)println("union(): " + union.collect().toList)//差集val subtract: RDD[Int] = rdd1.subtract(rdd2)println("subtract(): " + subtract.collect().toList)//交集val intersection: RDD[Int] = rdd1.intersection(rdd2)println("intersection(): " + intersection.collect().toList)//笛卡尔积val cartesian: RDD[(Int, Int)] = rdd1.cartesian(rdd2)println("cartesian(): " + cartesian.collect().toList)//zip 两个rdd的元素个数及分片个数必须是一样的，否则会报错val zip: RDD[(Int, Int)] = rdd1.zip(rdd2)println("zip(): " + zip.collect().toList)}
// 数据源1: List(1, 2, 3, 4)
// 数据源2: List(2, 3, 4, 5)
// union(): List(1, 2, 3, 4, 2, 3, 4, 5)
// subtract(): List(1)
// intersection(): List(2, 3, 4)
// cartesian(): List((1,2), (1,3), (1,4), (1,5), (2,2), (2,3), (2,4), (2,5), (3,2), (3,3), (3,4), (3,5), (4,2), (4,3), (4,4), (4,5))
// zip(): List((1,2), (2,3), (3,4), (4,5))

Key-Value类型交互

groupByKey()

作用：groupByKey也是对每个key进行操作，但只生成一个sequence

    val sc: SparkContext = spark.sparkContextval rdd1: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 1), ("AA", 13), ("BB", 2), ("BB", 5), ("CC", 7)))//分组之后求key对应的数量 指定分片数量之后输出的顺序或会改变val groupRdd: RDD[(String, Iterable[Int])] = rdd1.groupByKey()//分组之后进行聚合val rdd2: RDD[(String, Int)] = groupRdd.map(x => (x._1, x._2.sum))//　可以使用分片参数val rdd21: RDD[(String, Int)] = rdd1.groupByKey(3).map(x => (x._1, x._2.sum))val rdd22: RDD[(String, Int)] = rdd1.groupByKey(new HashPartitioner(3)).map(x => (x._1, x._2.sum))println("分组求和：　" + rdd2.collect().toList)println("分组求和：　" + rdd21.collect().toList)println("分组求和：　" + rdd22.collect().toList)// 分组求和：　List((BB,7), (CC,7), (AA,14))
// 分组求和：　List((BB,7), (AA,14), (CC,7))
// 分组求和：　List((BB,7), (AA,14), (CC,7))

reduceByKey(func, [numTasks])

作用：在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置

    // 等同上一个计算的结果val rdd3: RDD[(String, Int)] = rdd1.reduceByKey((a, b) => (a + b))val rdd31: RDD[(String, Int)] = rdd1.reduceByKey((a, b) => (a + b), 3)val rdd32: RDD[(String, Int)] = rdd1.reduceByKey(new HashPartitioner(3), (a, b) => (a + b))println("根据key计算：" + rdd3.collect().toList)println("根据key计算：" + rdd31.collect().toList)println("根据key计算：" + rdd32.collect().toList)// 根据key计算：List((BB,7), (CC,7), (AA,14))
// 根据key计算：List((BB,7), (AA,14), (CC,7))
// 根据key计算：List((BB,7), (AA,14), (CC,7))
// 结果和上一个分组求和完全一致

reduceByKey和groupByKey的区别

reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，返回结果是RDD[k,v]

groupByKey是按照key进行分组，直接进行shuffle(只是进行分组)

reduceByKey性能较高优先使用

sortByKey([ascending], [numTasks])

作用：在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD

    // 按照key的倒序排列val rdd4: RDD[(String, Int)] = rdd1.sortByKey(false)println("根据key计算：" + rdd4.collect().toList)// 根据key计算：List((CC,7), (BB,2), (BB,5), (AA,1), (AA,13))

join(otherDataset, [numTasks])

作用：在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD

    val j1: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 1), ("BB", 1), ("CC", 1), ("DD", 1)))val j2: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 2), ("BB", 2), ("CC", 2)))val j3: RDD[(String, (Int, Int))] = j1.join(j2, 3)println("形成新元组操作： " + j3.collect().toList)
// 形成新元组操作： List((BB,(1,2)), (AA,(1,2)), (CC,(1,2)))

mapValues(fun)

作用：【只操作value】针对于(K,V)形式的类型只对V进行操作

val sc: SparkContext = spark.sparkContextval rdd1: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 1), ("AA", 13), ("BB", 2), ("BB", 5), ("CC", 7)))    val rdd5: RDD[(String, String)] = rdd1.mapValues(x => x + "->Value")
println("仅操作value： " + rdd5.collect().toList)//仅操作value： List((AA,1->Value), (AA,13->Value), (BB,2->Value), (BB,5->Value), (CC,7->Value))

combineByKey[C]

作用：对相同K，把V合并成一个集合

参数：(createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C)

（1）createCombiner: combineByKey() 会遍历分区中的所有元素，因此每个元素的键要么还没有遇到过，要么就和之前的某个元素的键相同。如果这是一个新的元素,combineByKey()会使用一个叫作createCombiner()的函数来创建那个键对应的累加器的初始值

（2）mergeValue: 如果这是一个在处理当前分区之前已经遇到的键，它会使用mergeValue()方法将该键的累加器对应的当前值与这个新的值进行合并

（3）mergeCombiners: 由于每个分区都是独立处理的，因此对于同一个键可以有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器，就需要使用用户提供的 mergeCombiners() 方法将各个分区的结果进行合并

    val sc: SparkContext = spark.sparkContextval rdd1: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 1), ("AA", 13), ("BB", 2), ("BB", 5), ("CC", 7)))val value: RDD[(String, (Int, Int))] = rdd1.combineByKey(i => (i, 1), (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1), (acc1: (Int, Int), acc2: (Int, Int))=> (acc1._1 + acc2._1, acc1._2 + acc2._2))println("结果： " + value.collect().toList)
//               List(key,(sum,count))
// key联合计算： List((BB,(7,2)), (CC,(7,1)), (AA,(14,2)))

aggregateByKey

作用：在kv对的RDD中，，按key将value进行分组合并，合并时，将每个value和初始值作为seq函数的参数，进行计算，返回的结果作为一个新的kv对，然后再将结果按照key进行合并，最后将每个分组的value传递给combine函数进行计算（先将前两个value进行计算，将返回结果和下一个value传给combine函数，以此类推），将key与计算结果作为一个新的kv对输出

参数：(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U)

（1）zeroValue：给每一个分区中的每一个key一个初始值；

（2）seqOp：函数用于在每一个分区中用初始值逐步迭代value；

（3）combOp：函数用于合并每个分区中的结果。

foldByKey

作用：aggregateByKey的简化操作，seqop和combop相同

    val sc: SparkContext = spark.sparkContextval rdd1: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 1), ("AA", 13), ("BB", 2), ("BB", 10), ("BB", 5), ("CC", 7)))val rdd6: RDD[(String, Int)] = rdd1.aggregateByKey(0)((x, y) => math.max(x, y), (x, y) => x + y)println("key合计算： " + rdd6.collect().toList)val rdd7: RDD[(String, Int)] = rdd1.foldByKey(0)((x, y) => x + y)println("key合计算： " + rdd7.collect().toList)// key合计算： List((BB,17), (CC,7), (AA,14))

cogroup(otherDataset, [numTasks])

作用：在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

val sc: SparkContext = spark.sparkContext
val j1: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 1), ("BB", 1), ("BB", 3), ("CC", 1), ("DD", 1)))
val j2: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 2), ("BB", 2), ("CC", 2)))
val groupRdd1: RDD[(String, (Iterable[Int], Iterable[Int]))] = j1.cogroup(j2)
println("根据key进行分组:    " + groupRdd1.collect().toList)// 根据key进行分组:
//List((DD,(CompactBuffer(1),CompactBuffer())),
//     (BB,(CompactBuffer(1, 3),CompactBuffer(2))),
//     (CC,(CompactBuffer(1),CompactBuffer(2))),
//     (AA,(CompactBuffer(1),CompactBuffer(2))))

Action操作

PS: 此时才真正的执行rdd内部的计算和聚合

collect()

作用：在驱动程序中，以数组的形式返回数据集的所有元素

count()

作用：返回RDD中元素的个数

first()

作用：返回RDD中的第一个元素

take(n)

作用：返回一个由RDD的前n个元素组成的数组

takeOrdered(n)

作用：返回该RDD排序后的前n个元素组成的数组

countByKey()

作用：针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数

reduce(fun)

作用：通过func函数聚集RDD中的所有元素，先聚合分区内数据，再聚合分区间数据

    val sc: SparkContext = spark.sparkContextval rdd1: RDD[(String, Int)] = sc.makeRDD(Array(("AA", 1), ("BB", 5), ("BB", 3), ("CC", 7), ("DD", 2)))println("collect()： " + rdd1.collect().toList)println("count()： " + rdd1.count())println("first()： " + rdd1.first())println("take(3)： " + rdd1.take(3).toList)val countByKey: collection.Map[String, Long] = rdd1.countByKey()println("countByKey()： " + countByKey)// 输入的类型和输出的类型要一致val sum: Int = rdd1.values.reduce((x: Int, y: Int) => x + y)val sum1: (String, Int) = rdd1.reduce((x: (String, Int), y: (String, Int)) => (x._1 + "-" + y._1, x._2 + y._2))println("求和的结果： " + sum)println("求和的结果：+" + sum1)// 按照key的hashcode值为第一排序规则 value值倒序作为第二排序规则val takeOrdered: Array[(String, Int)] = rdd1.takeOrdered(3)(Ordering[(Int, Int)].on((x) => (-x._1.hashCode, -x._2)))println("原数据takeOrdered(3)： " + takeOrdered.toList)//  collect()： List((AA,1), (BB,5), (BB,3), (CC,7), (DD,2))
//  count()： 5
//  first()： (AA,1)
//  take(3)： List((AA,1), (BB,5), (BB,3))
//  countByKey()： Map(DD -> 1, BB -> 2, CC -> 1, AA -> 1)
//  求和的结果： 18
//  求和的结果：+(DD-AA-BB-BB-CC,18)
//  原数据takeOrdered(3)： List((DD,2), (CC,7), (BB,5))

aggregate

作用：aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合，然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致

参数：(zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U)

fold(num)(func)

作用：折叠操作，aggregate的简化操作，seqop和combop一样

    //1,2,3,4,5。6.7.8.9.10val rdd2: RDD[Int] = sc.parallelize(1 to 10)val aggregate: Int = rdd2.aggregate(0)((x, y) => x + y, (x, y) => x + y)println("aggregate(0)： " + aggregate)val fold: Int = rdd2.fold(0)((x, y) => x + y)println("fold(0)： " + fold)//  输出// aggregate(0)： 55// fold(0)： 55

foreach(fun)

作用：在数据集的每一个元素上，运行函数func进行更新

    val rdd2: RDD[Int] = sc.parallelize(1 to 10)rdd2.foreach(x => print(x + ", "))// 1, 3, 7, 6, 8, 2, 4, 5, 9, 10,

saveAsTextFile(path)

作用：将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本

saveAsObjectFile(path)

作用：用于将RDD中的元素序列化成对象，存储到文件中

    val rdd2: RDD[Int] = sc.parallelize(1 to 10)rdd2.saveAsTextFile("test.txt")rdd2.saveAsObjectFile("obj.txt")
//  可以指定文件生成路径，当前是生成在项目的根目录

Spark SQL编程之RDD-RDD转换相关推荐

dataframe记录数_大数据系列之Spark SQL、DataFrame和RDD数据统计与可视化
Spark大数据分析中涉及到RDD.Data Frame和SparkSQL的操作,本文简要介绍三种方式在数据统计中的算子使用. 1.在IPython Notebook运行Python Spark程序 ...
ORACLE PL/SQL编程之二：PL/SQL块结构和组成元素
2.1 PL/SQL块 PL/SQL程序由三个块组成,即声明部分.执行部分.异常处理部分. PL/SQL块的结构如下: DECLARE --声明部分: 在此声明PL/SQL用到的变量,类型及游标,以及 ...
linux网路编程之TCP状态转换及端口复用
(1)TCP状态转换图其中图中分为三种状态:实线代表的主动发起连接,虚线代表的被动发起连接,细实线代表的可以双向发起连接的状态. 主动发起连接方状态变化:1)主动发起连接的一方发送SYN标志位,进入 ...
Spark SQL程序实现RDD转换DataFrame
通过反射推断Schema 在Spark SQL中有两种方式可以在DataFrame和RDD进行转换利用反射机制,推导包含某种类型的RDD,通过反射将其转换为指定类型的DataFrame,适用于提前知 ...
Spark SQL之RDD转DataFrame
准备文件首先准备好测试文件info.txt,内容如下: 1,vincent,20 2,sarah,19 3,sofia,29 4,monica,26 将RDD转成DataFrame 方式一:反射可 ...
spark sql 上个月_Spark学习之路（十八）SparkSQL简单使用
一.SparkSQL的进化之路 1.0以前: Shark 1.1.x开始: SparkSQL(只是测试性的) SQL 1.3.x: SparkSQL(正式版本)+Dataframe 1.5.x: S ...
Spark SQL: Relational Data Processing in Spark
Spark SQL: Relational Data Processing in Spark Spark SQL : Spark中关系型处理模块说明: 类似这样的说明并非是原作者的内容翻译,而是本篇 ...
大数据Hadoop之——Spark SQL+Spark Streaming
文章目录一.Spark SQL概述二.SparkSQL版本 1)SparkSQL的演变之路 2)shark与SparkSQL对比 3)SparkSession 三.RDD.DataFrames和D ...
Spark SQL概述
1.介绍 Spark SQL是Spark用来处理结构化数据的一个模块,它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用.使用Spark SQL有两种方式,包括SQL语句以及 ...

Spark SQL编程之RDD-RDD转换

背景

备注

RDD创建

创建方式

RDD转换

Value类型交互

map(fun)

mapPartitions(fun)

mapPartitionsWithIndex(fun)

flatMap(fun)

glom()

groupBy(fun)

filter(fun)

distinct([numTasks]))

sortBy(func,[ascending], [numTasks])

coalesce(numPartitions)

repartition(numPartitions)

sample(withReplacement, fraction, seed)

pipe(command, [envVars])

双Value类型交互

union(otherDataset)

subtract (otherDataset)

intersection(otherDataset)

zip(otherDataset)

cartesian(otherDataset)

Key-Value类型交互

groupByKey()

reduceByKey(func, [numTasks])

sortByKey([ascending], [numTasks])

join(otherDataset, [numTasks])

mapValues(fun)

combineByKey[C]

aggregateByKey

foldByKey

cogroup(otherDataset, [numTasks])

Action操作

collect()

count()

first()

take(n)

takeOrdered(n)

countByKey()

reduce(fun)

aggregate

fold(num)(func)

foreach(fun)

saveAsTextFile(path)

saveAsObjectFile(path)

Spark SQL编程之RDD-RDD转换相关推荐

最新文章

热门文章