sparksql作为内存

[1]参考文章: 高彦杰，陈冠成Spark SQL: 基于内存的大数据分析引擎“程序员” 2014. 8

AMPLab将大数据分析工作负载分为三种类型: 批处理数据处理，交互式查询和实时流处理. 最重要的部分是交互式查询. 大数据分析堆栈需要满足用户的即席，报表，迭代和其他类型的查询需求. 它还需要提供一个SQL接口，以与原始用户的使用习惯兼容. 它还要求SQL能够重组关系模式. 完成这些重要的SQL任务的是两个开源的分布式大数据查询引擎，Spark SQL和Shark，它们可以理解为在Spark上实现轻量级Hive SQL的实现. 行业统称为Hadoop上的SQL之类的技术.

在2014年Spark峰会上，Databricks宣布将不再支持Shark开发，并将全力以赴开发Shark的下一代技术Spark SQL. 同时，Hive社区还启动了Hive onSpark项目，使用Spark作为Hive（MapReduce和Tez除外）New执行引擎. 根据伯克利的BigData Benchmark测试比较数据，Shark的In Memory性能可以达到Hive的100倍，甚至On Disk也可以实现10倍的性能提升，这是Hive的强大替代解决方案. 在Shark的最新版本中，Spark SQL在AMPLab的最新测试中胜过Shark. 图3-1显示了Spark SQL和Hive on Spark是新的开发方向.

开发……到Spark SQL: Shark开发已终止，请转到Spark SQL

Spark的新功能……: 基于Spark的新SQL查询引擎

帮助……到Spark: 帮助现有的Hive用户迁移到Spark

图3-1 Spark SQL和Hive on Spark是新的开发方向

由于Shark的底层依赖于Hive，因此该体系结构的优势在于，传统的Hive用户可以将Shark无缝集成到现有系统中以运行查询负载. 但是我们也看到一些问题: 随着版本的升级，查询优化器依赖于Hive，添加新的优化策略很不方便，并且需要另一个学习和二次开发系统，并且学习成本很高. 另一方面，MapReduce是进程级并行性. 例如，Hive将在不同的过程空间中使用一些静态变量. 当多个线程在同一进程空间中并行执行时，多个线程同时写入具有相同名称的静态变量将导致一致性问题. 因此，Shark需要使用另一套独立维护的Hive源分支. 为了解决这个问题，AMPLab和Databricks使用Catalyst开发了SparkSQL.

Spark的全栈解决方案为用户提供了各种数据分析框架. 机器学习，图形计算和流计算的发展和普及吸引了众多学习者. 为什么我们今天仍然关注大数据环境？要使用SQL？作者认为主要有以下原因:

1）易于使用和用户惯性. 在过去的多年中，由于SQL的易用性提高了应用程序开发的效率，因此许多程序员围绕+应用程序体系结构完成了工作. 程序员已经习惯了调用SQL编写程序的业务逻辑代码模式. 惯性的力量是强大的. 如果您可以按原始方式解决现有的大数据问题，为什么不这样做呢？提供SQL和JDBC支持将使传统用户能够像以前一样编写程序，从而大大降低了迁移成本.

2）生态系统的力量. 许多系统具有良好的软件性能，但是它们却不成功且在下降，主要是由于生态系统问题. 传统的SQL在JDBC，ODBC和SQL的各种标准下已经形成了成熟的生态系统. 可以迁移和使用许多应用程序组件和工具. 像某些可视化工具，数据分析工具等一样，原始的企业IT工具也可以无缝过渡.

3）数据去耦. Spark SQL正在扩展以支持多个持久层. 用户可以使用原始的持久层来存储数据，但他们也可以体验并迁移到Spark SQL为大数据提供的大数据. 分析.

Spark SQL与传统的DBMS查询优化器+执行程序体系结构更相似，不同之处在于其执行程序在分布式环境中实现，并使用Spark作为执行引擎. SparkSQL的查询优化是Catalyst. 它是基于Scala语言开发的. 它可以灵活地使用Scala的本地语言功能来促进功能扩展，并为Spark SQL的开发留出空间. Catalyst将SQL语言转换为最终执行计划，并在此过程中执行查询优化. 与传统的区别在于，SQL最终由查询优化器转换为可执行查询计划，该查询优化器是一个查询树，而传统的DB可以执行此查询计划. Spark SQL的最终执行仍会将该执行计划树转换为Spark中Spark的有向无环图DAG，然后执行它.

下面我们可以看到Catalyst的总体架构:

阶段……相关查询: 规则分析，优化和查询计划的各个阶段

未解决的逻辑计划: 未解决的逻辑计划

逻辑计划: 逻辑查询计划

优化的逻辑计划: 优化的逻辑查询计划

物理计划: 物理查询计划

分析规则: 分析规则

优化规则: 优化规则

计划规则: 计划规则

图3-2 Spark SQL查询引擎Catalyst的体系结构

从图3-2可以看出，整个Catalyst是Spark SQL的调度核心. 遵循传统的查询解析步骤，对SQL进行解析，将其转换为逻辑查询计划创建内存，物理查询计划，最后转换为Spark. 执行DAG. 图3-3显示了Catalyst的执行流程.

图3-3 Catalyst执行流程

SQlParser将SQL语句转换为逻辑查询计划. 分析器对逻辑查询计划执行属性和关系关联测试. 然后，Optimizer通过逻辑查询优化将逻辑查询计划转换为优化的逻辑查询计划. QueryPlanner转换优化的逻辑查询计划. 转换为物理查询计划后，prepareForExecution调整了数据分布，最后将物理查询计划转换为执行计划以输入Spark执行任务.

查询优化是传统中最重要的链接. 这项技术在传统中已经非常成熟. 除了查询优化之外，Spark SQL还针对存储进行了优化. 让我们从以下几点来看一下Spark SQL的一些优化策略:

（1）内存列存储和内存缓存表

Spark SQL可以使用cacheTable将数据存储转换为列存储并将数据加载到内存以进行缓存. cacheTable等效于分布式集群中内存的实例化视图，该视图缓存数据，因此迭代或交互式查询不再需要从HDFS读取数据，直接从内存读取数据大大减少了I / O开销. 列式存储的优势在于，Spark SQL只需要读出用户所需的列，而不是像行存储一样一次读出所有列，这大大减少了内存缓存数据量，更有效地使用了内存数据缓存. 减少网络传输和IO开销. 数据存储在列中. 由于相同数据类型的数据可以连续存储，因此可以使用序列化和压缩来减少内存空间的占用.

（2）列存储压缩

为了减少内存和硬盘空间的占用，Spark SQL使用一些压缩策略来压缩内存列存储数据. Spark SQL的压缩方法比Shark丰富得多. 例如，它支持PassThrough，RunLengthEncoding，DictionaryEncoding，BooleanBitSet，IntDelta，LongDelta和其他压缩方法. 这样可以大大减少内存空间占用，网络传输开销和I / O开销.

（3）逻辑查询优化

SparkSQL支持逻辑查询优化方法，例如在逻辑查询优化上进行列修剪，断言和属性合并（请参见图3-4）. 列修剪为了减少不必要的属性列的读取，减少数据传输和计算开销，将在查询优化器的转换过程中对列修剪进行优化.

这里我们介绍一个逻辑优化的例子:

SELECT ClassFROM（SELECT ID，名称，来​​自学生的类）在S.ID = 1的地方

图3-4逻辑查询优化

Catalyst通过谓词下推原始查询，并选择要首先执行的操作ID = 1，以便对大多数数据进行过滤，并且最终投影仅由属性合并完成一次，而Class属性列保留.

（4）联接优化

Spark SQL深刻地借鉴了传统的查询优化技术的本质，并在分布式环境中进行了特定的优化策略调整和创新. 现在，Spark SQL已经优化了Join，以支持各种连接算法. 当前的连接算法已经比Shark丰富，并且许多原始的Shark元素已逐渐迁移. 例如: BroadcastHashJoin，BroadcastNestedLoopJoin，HashJoin创建内存，LeftSemiJoin等.

这里，我们介绍了BroadcastHashJoin算法思想之一: BroadcastHashJoin将小表转换为广播变量以进行广播，从而避免Shuffle开销，并最终在分区中建立Hash连接. 在Hive中使用Map Side Join的想法. 同时，使用DBMS中的Hash连接算法进行连接.

随着Spark SQL的发展，将来将添加更多查询优化策略. 同时，Spark SQL将支持Shark Server，JDBC接口等服务器，并与NoSQL，传统DBMS等持久性层兼容. 强大的结构化大数据查询引擎正在兴起.

3. 如何使用Spark SQL

val sqlContext =新的org.apache.spark.sql.SQLContext（sc）

//这里介绍sqlContext下的所有方法，我们可以直接使用sql方法进行查询.

导入sqlContext. _

案例类Person（名称: 字符串，年龄: 整数）

//以下人员是包含案例类型数据的RDD. Scala的隐式机制默认会将RDD转换为SchemaRDD. SchemaRDD是SparkSQL中的核心RDD.

val people = sc.textFile（“ examples / src / main / resources / people.txt”）. map（_. split（“，”））. map（p => Person（p（0），p （1）.trim.toInt））

//内存元数据中的注册表信息，例如Spark SQL表已创建.

people.registerAsTable（“ people”）

// sql语句将触发上面分析的SparkSQL的执行过程. 读者可以参考上面的插图.

val青少年= sql（“从年龄> = 13 AND年龄<= 19”的人中选择姓名

///最后一代的青少年也是RDD

teenagers.map（t =>“名称: ” + t（0））. collect（）. foreach（println）

通过前面的介绍，读者对支持结构化数据分析任务的Spark SQL的原理和用法有了一定的了解. 在生产环境中，存在一种类型的数据分析任务，它对响应延迟有很高的要求，并且需要实时处理流数据. 在BDAS中，Spark Streaming用于支持流处理分析任务.

友好推荐: ABC技术研究所

为技术人员创建的专有A（AI），B（大数据），C（云）技术公共帐户和技术通信社区.

本文来自电脑杂谈，转载请注明本文网址：
https://www.pc-fly.com/a/jisuanjixue/article-249711-1.html