文档首页/ MapReduce服务 MRS/ 开发指南（LTS版）/ Spark2x开发指南（普通模式）/ Spark应用开发常见问题/ Spark常用API介绍/ Spark scala API接口介绍
更新时间：2024-10-31 GMT+08:00
查看PDF
分享

Spark scala API接口介绍

由于Spark开源版本升级，为避免出现API兼容性或可靠性问题，建议用户使用配套版本的API。

Spark Core常用接口

Spark主要使用到如下这几个类：

  • SparkContext：是Spark的对外接口，负责向调用该类的scala应用提供Spark的各种功能，如连接Spark集群，创建RDD等。
  • SparkConf：Spark应用配置类，如设置应用名称，执行模式，executor内存等。
  • RDD（Resilient Distributed Dataset）：用于在Spark应用程序中定义RDD的类，该类提供数据集的操作方法，如map，filter。
  • PairRDDFunctions：为key-value对的RDD数据提供运算操作，如groupByKey。
  • Broadcast：广播变量类。广播变量允许保留一个只读的变量，缓存在每一台机器上，而非每个任务保存一份复制。
  • StorageLevel：数据存储级别。有内存（MEMORY_ONLY），磁盘（DISK_ONLY），内存+磁盘（MEMORY_AND_DISK）等。
RDD上支持两种类型的操作：Transformation和Action，这两种类型的常用方法如表1和表2所示。
表1 Transformation

方法

说明

map[U](f: (T) => U): RDD[U]

对调用map的RDD数据集中的每个element都使用f方法，生成新的RDD。

filter(f: (T) => Boolean): RDD[T]

对RDD中所有元素调用f方法，生成将满足条件数据集以RDD形式返回。

flatMap[U](f: (T) => TraversableOnce[U])(implicit arg0: ClassTag[U]): RDD[U]

先对RDD所有元素调用f方法，然后将结果扁平化，生成新的RDD。

sample(withReplacement: Boolean, fraction: Double, seed: Long = Utils.random.nextLong): RDD[T]

抽样，返回RDD一个子集。

union(other: RDD[T]): RDD[T]

返回一个新的RDD，包含源RDD和给定RDD的元素的集合。

distinct([numPartitions: Int]): RDD[T]

去除重复元素，生成新的RDD。

groupByKey(): RDD[(K, Iterable[V])]

返回(K,Iterable[V])，将key相同的value组成一个集合。

reduceByKey(func: (V, V) => V[, numPartitions: Int]): RDD[(K, V)]

对key相同的value调用func。

sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]

按照key来进行排序，是升序还是降序，ascending是boolean类型。

join[W](other: RDD[(K, W)][, numPartitions: Int]): RDD[(K, (V, W))]

当有两个KV的dataset(K,V)和(K,W)，返回的是(K,(V,W))的dataset，numPartitions为并发的任务数。

cogroup[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Iterable[V], Iterable[W]))]

将当有两个key-value对的dataset(K,V)和(K,W)，返回的是(K, (Iterable[V], Iterable[W]))的dataset，numPartitions为并发的任务数。

cartesian[U](other: RDD[U])(implicit arg0: ClassTag[U]): RDD[(T, U)]

返回该RDD与其它RDD的笛卡尔积。
表2 Action

API

说明

reduce(f: (T, T) => T):

对RDD中的元素调用f。

collect(): Array[T]

返回包含RDD中所有元素的一个数组。

count(): Long

返回的是dataset中的element的个数。

first(): T

返回的是dataset中的第一个元素。

take(num: Int): Array[T]

返回前n个elements。

takeSample(withReplacement: Boolean, num: Int, seed: Long = Utils.random.nextLong): Array[T]

takeSample(withReplacement，num，seed)对dataset随机抽样，返回由num个元素组成的数组。withReplacement表示是否使用replacement。

saveAsTextFile(path: String): Unit

把dataset写到一个text file、HDFS或者HDFS支持的文件系统中，spark把每条记录都转换为一行记录，然后写到file中。

saveAsSequenceFile(path: String, codec: Option[Class[_ <: CompressionCodec]] = None): Unit

只能用在key-value对上，然后生成SequenceFile写到本地或者hadoop文件系统。

countByKey(): Map[K, Long]

对每个key出现的次数做统计。

foreach(func: (T) => Unit): Unit

在数据集的每一个元素上，运行函数func。

countByValue()(implicit ord: Ordering[T] = null): Map[T, Long]

对RDD中每个元素出现的次数进行统计。
表3 Spark Core新增接口

API

说明

isSparkContextDown:AtomicBoolean

该接口可判断sparkContext是否已完全stop，初始值为false。

若接口值为true，则代表sparkContext已完全stop。

若接口值为false，则代表sparkContext没有完成stop。

例如：用户根据 sc.isSparkContextDown.get() == true 可判断sparkContext已完全stop。

Spark Streaming常用接口

Spark Streaming中常见的类有：

  • StreamingContext：是Spark Streaming功能的主入口，负责提供创建DStreams的方法，入参中需要设置批次的时间间隔。
  • dstream.DStream：是一种代表RDDs连续序列的数据类型，代表连续数据流。
  • dstream.PariDStreamFunctions：键值对的DStream，常见的操作如groupByKey和reduceByKey。

    对应的Spark Streaming的JAVA API是JavaStreamingContext，JavaDStream和JavaPairDStream。

Spark Streaming的常见方法与Spark Core类似，下表罗列了Spark Streaming特有的一些方法。

表4 Spark Streaming方法介绍

方法

说明

socketTextStream(hostname: String, port: Int, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2): ReceiverInputDStream[String]

从TCP源主机：端口创建一个输入流。

start():Unit

启动Spark Streaming计算。

awaitTermination(timeout: long):Unit

当前进程等待终止，如Ctrl+C等。

stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit

终止Spark Streaming计算。

transform[T](dstreams: Seq[DStream[_]], transformFunc: (Seq[RDD[_]], Time) ? RDD[T])(implicit arg0: ClassTag[T]): DStream[T]

对每一个RDD应用function操作得到一个新的DStream。

UpdateStateByKey(func)

更新DStream的状态。使用此方法，需要定义状态和状态更新函数。

window(windowLength, slideInterval)

根据源DStream的窗口批次计算得到一个新的DStream。

countByWindow(windowLength, slideInterval)

返回流中滑动窗口元素的个数。

reduceByWindow(func, windowLength, slideInterval)

当调用在DStream的KV对上，返回一个新的DStream的KV对，其中每个Key的Value根据滑动窗口中批次的reduce函数聚合得到。

join(otherStream, [numTasks])

实现不同的Spark Streaming之间做合并操作。

DStreamKafkaWriter.writeToKafka()

支持将DStream中的数据批量写入到Kafka。

DStreamKafkaWriter.writeToKafkaBySingle()

支持将DStream中的数据逐条写入到Kafka。
表5 Spark Streaming增强特性接口

方法

说明

DStreamKafkaWriter.writeToKafka()

支持将DStream中的数据批量写入到Kafka。

DStreamKafkaWriter.writeToKafkaBySingle()

支持将DStream中的数据逐条写入到Kafka。

SparkSQL常用接口

Spark SQL中常用的类有：

  • SQLContext：是Spark SQL功能和DataFrame的主入口。
  • DataFrame：是一个以命名列方式组织的分布式数据集。
  • HiveContext：获取存储在Hive中数据的主入口。
表6 常用的Actions方法

方法

说明

collect(): Array[Row]

返回一个数组，包含DataFrame的所有列。

count(): Long

返回DataFrame中的行数。

describe(cols: String*): DataFrame

计算统计信息，包含计数，平均值，标准差，最小值和最大值。

first(): Row

返回第一行。

Head(n:Int): Row

返回前n行。

show(numRows: Int, truncate: Boolean): Unit

用表格形式显示DataFrame。

take(n:Int): Array[Row]

返回DataFrame中的前n行。
表7 基本的DataFrame Functions

方法

说明

explain(): Unit

打印出SQL语句的逻辑计划和物理计划。

printSchema(): Unit

打印schema信息到控制台。

registerTempTable(tableName: String): Unit

将DataFrame注册为一张临时表，其周期和SQLContext绑定在一起。

toDF(colNames: String*): DataFrame

返回一个列重命名的DataFrame。
父主题： Spark常用API介绍