Spark基础-低级非结构化API
低级API
有两种低级API
处理分布式数据（RDD）
用于分发和处理分布式共享变量（广播变量和累加器）
使用时机
当高级API不满足需求时，比如对集群中数据的物理存储进行非常严格的操作
需要维护遗留的旧版本代码时
需要执行一些自定义共享变量操作时
RDD(弹性分布式数据集)
RDD是一个只读不可变的且已经分块的记录合集，可以被并行处理。RDD中记录的仅仅是对应的Java，scala或python对象。可以以任何格式在这些对象中存储任何内容。
RDD类型
每个RDD包含以下5个主要内部属性
- 数据分片列表
- 作用在每个数据分片的计算函数
- 描述与其他RDD的依赖关系列表
- （可选）为key-value RDD配置的分片方法
- （可选）优先位置列表，根据本地数据特性，指定了每个分片的处理位置偏好（如HDFS文件，这个列表就是每个文件块所在的节点）
创建RDD
利用df或者ds的rdd方法就能转换为rdd
RDD也可以转换会df或ds
通过spark上下文读取文件，读取结果将生成一个RDD
操作RDD
RDD操作的是原始的Java/Scala对象而非Spark类型，还缺乏辅助方法和函数。因此所有的filter，map等方法都必须自定义。
比如定义一个filter操作：
保存文件
Spark会把RDD中每个分区都读出来，并写到指定位置中。
mapPartitions
mapPartitions函数每次处理一个数据分区（可以通过迭代器遍历这个分区的数据）。
foreachpartitions
foreachpartitions没有返回值，仅用于迭代所有数据分区，用于写入数据库等不需要返回值的操作。
高级RDD操作
key-value
许多方法都是基于key-value格式的RDD进行的，方法名中包含ByKey就意味着只能以pairRDD的类型进行操作。
通过keyBy函数，将普通rdd转换成kv形式
聚合操作
对普通RDD和kvRDD都可以进行聚合操作，有相应的groupByKey，reduceByKey。还有groupBy和reduceBy方法。
着重强调：不要使用groupByKey方法！！！！
这个方法需要在内存中保存一个key对应的所有value，然后进行计算。如果有严重的key倾斜现象，会出现OOM问题。
正确的聚合应该使用reduceByKey方法，reduce方法的计算发生在每个分组，将计算的中间结果进行传递汇总，不会导致OOM，也不会有shuffle的过程。
注意：因为是RDD，所以使用的addFunc是自己实现的。
控制分区
与结构化API不同，RDD允许指定一个分区函数partitioner，来精准控制数据在整个集群上的物理分布。
结构化的API不支持自定义分区。RDD支持，其目的是将数据均匀分布到整个集群，避免数据倾斜等问题。
一般使用过程是将DS或DF降级为RDD，应用自定义分区程序，再转回DF或DS
具体实现如下：
首先读取数据，转换为RDD，之后rdd转换为KeyedRDD，再调用partitionBy方法。需要传入一个partitioner类。此处自己实现了一个（继承Partitioner）。spark内置了两个partitioner，分别是处理离散值的HashPartitioner和处理连续值的RangePartitioner
自定义序列化
任何希望并行处理（或函数操作）的对象都必须是可序列话的。即需要继承Serializable
默认的序列化方式可能很慢。spark可以使用kryo库更快的进行序列化。
具体做法如下：
- 设置spark配置spark.serializer 的值为 org.apache.spark.serializer.KryoSerializer
- 注册自定义类
分布式共享变量
分布式共享变量包括两种
广播变量是共享的，不可修改的。缓存在集群中的每个节点上，而不是在每个任务中反复序列化。
具体使用如下：
首先创建一个map的val，之后通过sparkContext的broadcast方法创建一个广播变量suppBroadcast。之后就可以在RDD的相关操作，此处举例为map中使用了。具体使用方式是通过.value拿到对应的值。
累加器
累加器提供一个累加用的变量，Spark集群可以按行方式对其进行安全更新。仅支持由满足交换律和结合律的操作进行累加的变量。因此累加器可以高效并行。
spark保证每个任务对累加器的更新只进行一次，重启的任务不会再更新。
具体使用如下：
首先创建一个累加器，可以实例化后注册，也可以直接通过context创建。
之后定义一个累加方法，其中需要调用累加器的add方法进行累加操作
最后在DF上调用累加方法，实现累加功能foreach中每一行都运行一次累加方法。