引言:
Spark是在借鉴了MapReduce之上发展而来的,继承了其分布式并行计算的优点并改进了MapReduce明显的缺陷。Spark主要包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX等组件。
本文主要分析了 Spark RDD 以及 RDD 作为开发的不足之处,介绍了 SparkSQL 对已有的常见数据系统的操作方法,以及重点介绍了普元在众多数据开发项目中总结的基于 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 提高很大的便利性,但在使用上仍然有许多问题。体现在一下几个方面:
三、SparkSQL
Spark 从 1.3 版本开始原有 SchemaRDD 的基础上提供了类似Pandas DataFrame API。新的DataFrame API不仅可以大幅度降低普通开发者的学习门槛,同时还支持Scala、Java与Python三种语言。更重要的是,由于脱胎自SchemaRDD,DataFrame天然适用于分布式大数据场景。
一般的数据处理步骤:读入数据 -> 对数据进行处理 -> 分析结果 -> 写入结果
SparkSQL 结构化数据
处理非结构化数据,不能简单的用 DataFrame 装载。而是要用 SparkRDD 把数据读入,在通过一系列的 Transformer Method 把非结构化的数据加工为结构化,或者过滤到不合法的数据。
SparkSQL DataFrame
SparkSQL 中一切都是 DataFrame,all in DataFrame. DataFrame是一种以RDD为基础的分布式数据集,类似于传统数据库中的二维表格。DataFrame与RDD的主要区别在于,前者带有schema元信息,即DataFrame所表示的二维表数据集的每一列都带有名称和类型。如果熟悉 Python Pandas 库中的 DataFrame 结构,则会对 SparkSQL DataFrame 概念非常熟悉。
TextFile DataFrame
- import.org.apache.spark.sql._
- //定义数据的列名称和类型
- valdt=StructType(List(id:String,name:String,gender:String,age:Int))
-
- //导入user_info.csv文件并指定分隔符
- vallines = sc.textFile("/path/user_info.csv").map(_.split(","))
-
- //将表结构和数据关联起来,把读入的数据user.csv映射成行,构成数据集
- valrowRDD = lines.map(x=>Row(x(0),x(1),x(2),x(3).toInt))
-
- //通过SparkSession.createDataFrame()创建表,并且数据表表头
- val df= spark.createDataFrame(rowRDD, dt)
读取规则数据文件作为DataFrame
- SparkSession.Builder builder = SparkSession.builder()
- Builder.setMaster("local").setAppName("TestSparkSQLApp")
- SparkSession spark = builder.getOrCreate();
- SQLContext sqlContext = spark.sqlContext();
-
- # 读取 JSON 数据,path 可为文件或者目录
- valdf=sqlContext.read().json(path);
-
- # 读取 HadoopParquet 文件
- vardf=sqlContext.read().parquet(path);
-
- # 读取 HadoopORC 文件
- vardf=sqlContext.read().orc(path);
JSON 文件为每行一个 JSON 对象的文件类型,行尾无须逗号。文件头也无须[]指定为数组;SparkSQL 读取是只是按照每行一条 JSON Record序列化;
Parquet文件
- Configurationconfig = new Configuration();
- ParquetFileReaderreader = ParquetFileReader.open(
- HadoopInputFile.fromPath(new Path("hdfs:///path/file.parquet"),conf));
- Map<String, String>schema = reader.getFileMetaData().getKeyValueMetaData();
- String allFields= schema.get("org.apache.spark.sql.parquet.row.metadata");
allFiedls 的值就是各字段的名称和具体的类型,整体是一个json格式进行展示。
读取 Hive 表作为 DataFrame
Spark2 API 推荐通过 SparkSession.Builder 的 Builder 模式创建 SparkContext。 Builder.getOrCreate() 用于创建 SparkSession,SparkSession 是 SparkContext 的封装。
在Spark1.6中有两个核心组件SQLcontext和HiveContext。SQLContext 用于处理在 SparkSQL 中动态注册的表,HiveContext 用于处理 Hive 中的表。
从Spark2.0以上的版本开始,spark是使用全新的SparkSession接口代替Spark1.6中的SQLcontext和HiveContext。SQLContext.sql 即可执行 Hive 中的表,也可执行内部注册的表;
在需要执行 Hive 表时,只需要在 SparkSession.Builder 中开启 Hive 支持即可(enableHiveSupport())。
- SparkSession.Builder builder = SparkSession.builder().enableHiveSupport();
- SparkSession spark = builder.getOrCreate();
- SQLContext sqlContext = spark.sqlContext();
// db 指 Hive 库中的数据库名,如果不写默认为 default
// tableName 指 hive 库的数据表名
- sqlContext.sql(“select * from db.tableName”)
SparkSQL ThriftServer
//首先打开 Hive 的 Metastore服务
- hive$bin/hive –-service metastore –p 8093
//把 Spark 的相关 jar 上传到hadoophdfs指定目录,用于指定sparkonyarn的依赖 jar
- spark$hadoop fs –put jars/*.jar /lib/spark2
// 启动 spark thriftserver 服务
- spark$ sbin/start-thriftserver.sh --master yarn-client --driver-memory 1G --conf
- spark.yarn.jars=hdfs:///lib/spark2/*.jar
当hdfs 上传了spark 依赖 jar 时,通过spark.yarn.jars 可看到日志 spark 无须每个job 都上传jar,可节省启动时间
- 19/06/1114:08:26 INFO Client: Source and destination file systems are the same. Notcopying hdfs://localhost:9000/lib/spark2/snappy-java-1.0.5.jar
- 19/06/1114:08:26 INFO Client: Source and destination file systems are the same. Notcopying hdfs://localhost:9000/lib/spark2/snappy-java-1.1.7.3.jar
//通过 spark bin 下的 beeline 工具,可以连接到 spark ThriftServer(SparkOnHive)
- bin/beeline -u jdbc:hive2://ip:10000/default -n hadoop
Beeline 还支持传入-e 可传入一行 SQL,
也可通过 –f 指定一个 SQL File,内部可用逗号分隔的多个 SQL(存储过程)
SparkSQL Beeline 的执行效果展示
SparkSQL ThriftServer
对于 SparkSQL ThriftServer 服务,每个登陆的用户都有创建的 SparkSession,并且执行的对个 SQL 会通过时间顺序列表展示。
SparkSQL ThriftServer 服务可用于其他支持的数据库工具创建查询,也用于第三方的 BI 工具,如 tableau。
四、SparkSQL Flow
SparkSQL Flow 是以 SparkSQL 为基础,开发的统一的基于 XML 配置化的可执行一连串的 SQL 操作,这一连串的 SQL 操作定义为一个 Flow。下文开始 SparkSQL Flow 的介绍:
SparkSQL Flow 是基于 SparkSQL 开发的一种基于 XML 配置化的 SQL 数据流转处理模型。该模型简化了 SparkSQL 、Spark RDD的开发,并且降低开发了难度,适合了解数据业务但无法驾驭大数据以及 Spark 技术的开发者。
SparkSQL Flow 适合的场景:
SparkSQL Flow XML 概览
如你所见,source 的 type 参数用于区分 source 的类型,source 支持的种类直接决定SparkSQL Flow 的数据源加载广度;并且,根据 type 不同,source 也需要配置不同的参数,如数据库还需要 driver,url,user和 password 参数。
Transformer 是基于 source 定的数据视图可执行的一组转换 SQL,该 SQL 符合 SparkSQL 的语法(SQL99)。Transform 的 SQL 的执行结果被作为中间表命名为 table_name 指定的值。
Targets 为定义输出,table_name 的值需在 source 或者 Transformer 中定义。
SparkSQL Flow 支持的Sourse
SparkSQL Flow TextFile Source
textfile 为读取文本文件,把文本文件每行按照 delimiter 指定的字符进行切分,切分不够的列使用 null 填充。
- <source type="textfile" table_name="et_rel_pty_cong"
- fields="cust_id,name1,gender1,age1:int"
- delimiter=","
- path="file:///Users/zhenqin/software/hive/user.txt"/>
SparkSQL Flow DB Source
- <source type="mysql" table_name="et_rel_pty_cong"
- table="user"
- url="jdbc:mysql://localhost:3306/tdb?characterEncoding=UTF-8"
- driver="com.mysql.jdbc.Driver"
- user="root" password="123456"/>
RDBMS 是从数据库使用 JDBC读取 数据集。支持 type 为:db、mysql、oracle、postgres、mssql;
SparkSQL Flow Transformer
- <transform type="sql" table_name="cust_id_agmt_id_t" cached="true">
- SELECT c_phone,c_type,c_num, CONCAT_VAL(cust_id) as cust_ids
- FROM user_concat_testx
- group by c_phone,c_type,c_num
- </transform>
Transform 支持 cached 属性,默认为 false;如果设置为 true,相当于把该结果缓存到内存中,缓存到内存中的数据在后续其它 Transform 中使用能提高计算效率。但是需使用大量内存,开发者需要评估该数据集能否放到内存中,防止出现 OutofMemory 的异常。
SparkSQL Flow Targets
SparkSQL Flow Targets 支持输出数据到一个或者多个目标。这些目标,基本覆盖了 Source 包含的外部系统。下面以 Hive 举例说明:
- <target type="hive"
- table_name="cust_id_agmt_id_t"
- savemode=”append”
- target_table_name="cust_id_agmt_id_h"/>
Target 有一个特殊的 show 类型的 target。用于直接在控制台输出一个 DataFrame 的结果到控制台(print),该 target 用于开发和测试。
- <target type="show" table_name="cust_id_agmt_id_t" rows=”10000”/>
Rows 用于控制输出多少行数据。
SparkSQL Around
After 用于 Flow 在运行结束后执行的一个环绕,用于记录日志和写入状态。类似 Java 的 try {} finally{ round.execute() }
多个 round 一定会执行,round 异常不会导致任务失败。
- <prepare>
- <round type="mysql"
- sql="insert into cpic_task_history(id, task_type, catalog_model, start_time, retry_count, final_status, created_at)
- values(${uuid}, ${task.type}, ${catalog.model}, ${starttime}, 0, ${status}, now())"
- url="${jdbc.url}" .../>
- </prepare>
- <after>
- <round type="mysql"
- sql="update cpic_task_history set
- end_time = ${endtime}, final_status = ${status}, error_text = ${error} where id = ${uuid}"
- url="${jdbc.url}”…/>
- </after>
Prepare round 和 after round 配合使用可用于记录 SparkSQL Flow 任务的运行日志。
SparkSQL Around的执行效果
Prepare round 可做插入(insert)动作,after round 可做更新 (update)动作,相当于在数据库表中从执行开始到结束有了完整的日志记录。SparkSQL Flow 会保证round 一定能被执行,而且 round 的执行不影响任务的状态。
SparkSQL Flow 提交
- bin/spark-submit --master yarn-client --driver-memory 1G
- --num-executors 10 --executor-memory 2G
- --jars /lib/jsoup-1.11.3.jarlib/jsqlparser-0.9.6.jar,/lib/mysql-connector-java-5.1.46.jar
- --conf spark.yarn.jars=hdfs:///lib/spark2/*.jar
- --queue default --name FlowTest
- etl-flow-0.2.0.jar -f hive-flow-test.xml
接收必须的参数 –f,可选的参数为支持 Kerberos 认证的租户名称principal,和其认证需要的密钥文件。
- usage: spark-submit --jars etl-flow.jar --class
- com.yiidata.etl.flow.source.FlowRunner
- -f,--xml-file <arg> Flow XML File Path
- --keytabFile <arg> keytab File Path(Huawei)
- --krb5File <arg> krb5 File Path(Huawei)
- --principal <arg> principal for hadoop(Huawei)
SparkSQL Execution Plan
每个Spark Flow 任务本质上是一连串的 SparkSQL 操作,在 SparkUI SQL tab 里可以看到 flow 中重要的数据表操作。
regiserDataFrameAsTable 是每个 source 和 Transform 的数据在 SparkSQL 中的数据视图,每个视图都会在 SparkContex 中注册一次。
对RegisterDataFrameAsTable的分析
通过单个 regiserDataFrameAsTable 项进行分析,SparkSQL 并不是把source 的数据立即计算把数据放到内存,而是每次执行 source 时只是生成了一个 Logical Plan,只有遇到需要提交的算子(Action),SparkSQL 才会触发前面所依赖的的 plan 执行。
总结
这是一个开发框架,不是一个成熟的产品,也不是一种架构。他只是基于 SparkSQL 整合了大多数的外部系统,能通过 XML 的模板配置完成数据开发。面向的是理解数据业务但不了解 Spark 的数据开发人员。整个框架完成了大多数的外部系统对接,开发者只需要使用 type 获得数据,完成数据开发后通过 target 回写到目标系统中。整个过程基本无须程序开发,除非当前的 SQL 函数无法满足使用的情况下,需要自行开发一下特定的 UDF。因此本框架在对 SparkSQL 做了二次开发基础上,大大简化了 Spark 的开发,可降低了开发者使用难度。
关于作者:震秦,普元资深开发工程师,专注于大数据开发 8 年,擅长 Hadoop 生态内各工具的使用和优化。参与某公关广告(上市)公司DMP 建设,负责数据分层设计和批处理,调度实现,完成交付使用;参与国内多省市公安社交网络项目部署,负责产品开发(Spark 分析应用);参与数据清洗加工为我方主题库并部署上层应用。
关于EAWorld:微服务,DevOps,数据治理,移动架构原创技术分享。
中国最?好的一朵云飘进了华瑞银行。阿里云将进一步助力华瑞银行All in Cloud。 -...
查看表结构,sbtest1有主键、k_1二级索引、i_c二级索引 CREATE TABLE `sbtest1` ...
本文转载自网络,原文链接:https://mp.weixin.qq.com/s/vlOUg46B5bcmToX-fjavJQ...
一、PostgreSQL行业位置 一 行业位置 首先我们看一看RDS PostgreSQL在整个行业当...
最近,DevOps的采用导致了企业计算的重大转变。除无服务器计算,动态配置和即付...
定义 this是函数运行时自动生成的内部对象,即调用函数的那个对象。(不一定很准...
很长时间没有更新原创文章了,但是还一直在思考和沉淀当中,后面公众号会更频繁...
在TOP云(zuntop.com)科技租赁过服务器的站长都知道独立服务器在价格上比VPS主...
9月17日,2020云栖大会上,阿里云正式发布工业大脑3.0。 阿里云智能资深产品专家...
2020年对于云计算行业来说是突破性的一年,因为公共云供应商增加了收入,而疫情...