基于Spark的数据分析实践

引言：

Spark是在借鉴了MapReduce之上发展而来的，继承了其分布式并行计算的优点并改进了MapReduce明显的缺陷。Spark主要包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX等组件。

本文主要分析了 Spark RDD 以及 RDD 作为开发的不足之处，介绍了 SparkSQL 对已有的常见数据系统的操作方法，以及重点介绍了普元在众多数据开发项目中总结的基于 SparkSQL Flow 开发框架。

目录：

1. Spark RDD
2. 基于Spark RDD数据开发的不足
3. SparkSQL
4. SparkSQL Flow

一、Spark RDD

RDD(Resilient Distributed Dataset)叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、元素可并行计算的集合。

RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。

//Scala 在内存中使用列表创建

val lines = List(“A”, “B”, “C”, “D” …) 
val rdd:RDD = sc.parallelize(lines); 

//以文本文件创建

val rdd:RDD[String] = sc.textFile(“hdfs://path/filename”) 

Spark RDD Partition 分区划分

新版本的 Hadoop 已经把 BlockSize 改为 128M，也就是说每个分区处理的数据量更大。

Spark 读取文件分区的核心原理

本质上，Spark 是利用了 Hadoop 的底层对数据进行分区的 API(InputFormat)：

public abstract class InputFormat<K,V>{ 
 public abstract List<InputSplit> getSplits(JobContextcontext 
 ) throwsIOException,InterruptedException; 
  
 public abstract RecordReader<K,V> createRecordReader(InputSplitsplit, 
 TaskAttemptContextcontext 
 )throwsIOException,InterruptedException; 
} 

Spark 任务提交后通过对输入进行 Split，在 RDD 构造阶段，只是判断是否可 Split(如果参数异常一定在此阶段报出异常)，并且 Split 后每个 InputSplit 都是一个分区。只有在Action 算子提交后，才真正用 getSplits 返回的 InputSplit 通过 createRecordReader 获得每个 Partition 的连接。

然后通过 RecordReader 的 next() 遍历分区内的数据。

Spark RDD 转换函数和提交函数

Spark RDD 的众多函数可分为两大类Transformation 与 Action。Transformation 与 Action 的区别在于，对 RDD 进行 Transformation 并不会触发计算：Transformation 方法所产生的 RDD 对象只会记录住该 RDD 所依赖的 RDD 以及计算产生该 RDD 的数据的方式;只有在用户进行 Action 操作时，Spark 才会调度 RDD 计算任务，依次为各个 RDD 计算数据。这就是 Spark RDD 内函数的“懒加载”特性。

二、基于Spark RDD数据开发的不足

由于MapReduce的shuffle过程需写磁盘，比较影响性能;而Spark利用RDD技术，计算在内存中流式进行。另外 MapReduce计算框架(API)比较局限, 使用需要关注的参数众多，而Spark则是中间结果自动推断，通过对数据集上链式执行函数具备一定的灵活性。

即使 SparkRDD 相对于 MapReduce 提高很大的便利性，但在使用上仍然有许多问题。体现在一下几个方面：

1. RDD 函数众多，开发者不容易掌握，部分函数使用不当 shuffle时造成数据倾斜影响性能;
2. RDD 关注点仍然是Spark太底层的 API，基于 Spark RDD的开发是基于特定语言(Scala，Python，Java)的函数开发，无法以数据的视界来开发数据;
3. 对 RDD 转换算子函数内部分常量、变量、广播变量使用不当，会造成不可控的异常;
4. 对多种数据开发，需各自开发RDD的转换，样板代码较多，无法有效重利用;
5. 其它在运行期可能发生的异常。如：对象无法序列化等运行期才能发现的异常。

三、SparkSQL

Spark 从 1.3 版本开始原有 SchemaRDD 的基础上提供了类似Pandas DataFrame API。新的DataFrame API不仅可以大幅度降低普通开发者的学习门槛，同时还支持Scala、Java与Python三种语言。更重要的是，由于脱胎自SchemaRDD，DataFrame天然适用于分布式大数据场景。

一般的数据处理步骤：读入数据 -> 对数据进行处理 -> 分析结果 -> 写入结果

SparkSQL 结构化数据

• 处理结构化数据(如 CSV，JSON，Parquet 等);
• 把已经结构化数据抽象成 DataFrame (HiveTable);
• 非结构化数据通过 RDD.map.filter 转换成结构化进行处理;
• 按照列式数据库，只加载非结构化中可结构化的部分列(Hbase，MongoDB);

处理非结构化数据，不能简单的用 DataFrame 装载。而是要用 SparkRDD 把数据读入，在通过一系列的 Transformer Method 把非结构化的数据加工为结构化，或者过滤到不合法的数据。

SparkSQL DataFrame

SparkSQL 中一切都是 DataFrame，all in DataFrame. DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame与RDD的主要区别在于，前者带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。如果熟悉 Python Pandas 库中的 DataFrame 结构，则会对 SparkSQL DataFrame 概念非常熟悉。

（编辑：东莞站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!