上篇在总体上介绍了sparkSQL的运行架构及其基本实现方法（Tree和Rule的配合），也大致介绍了sparkSQL中涉及到的各个概念和组件。本篇将详细地介绍一下关键的一些概念和组件，由于hiveContext继承自sqlContext，关键的概念和组件类似，只不过后者针对hive的特性做了一些修正和重写，所以本篇就只介绍sqlContext的关键的概念和组件。概念：LogicalPlan组件：SqlParserAnalyzerOptimizerPlanner1：LogicalPlan在sparkSQL的运行架构中，LogicalPlan贯穿了大部分的过程，其中catalyst中的SqlParser、Analyzer、Optimizer都要对LogicalPlan进行操作。LogicalPlan的定义如下:abstract class LogicalPlan extends QueryPlan[LogicalPlan] { self: Product => case class Statistics( sizeInBytes: BigInt ) lazy val statistics: Statistics = { if (children.size == 0) { throw new UnsupportedOperati ...继续阅读 (47)

上篇在总体上介绍了sparkSQL的运行架构及其基本实现方法（Tree和Rule的配合），也大致介绍了sparkSQL中涉及到的各个概念和组件。本篇将详细地介绍一下关键的一些概念和组件，由于hiveContext继承自sqlContext，关键的概念和组件类似，只不过后者针对hive的特性做了一些修正和重写，所以本篇就只介绍sqlContext的关键的概念和组件。 概念： Logi ...继续阅读 (5)

经过一段时间授课，积累下不少的spark知识，想逐步汇总成资料，分享给小伙伴们。对于想视频学习的小伙伴，可以访问炼数成金网站的《spark大数据平台》课程，每周的课程是原理加实际操作。最新的课程是第3期，至于费用，越认真学习，学习成本越低。从Spark1.1.0开始，重新规划了一下学习路线，由于涉及的内容比较多，会不断的完善其中的内容，争取在Spark1.2.0形成一个完整的体系，最迟到Spark1.3.0。笔者计划在未来的一年时间内，将大部分的空闲时间都花在Spark学习路线的完善之上。在撰写的过程中，笔者将按照一个一个主题来编写博客，写完一个主题后汇编成PDF进行分享。最新的分享是《SparkSQL1.1.0入门》，所有分享的汇总文档都将放在百度网盘上。下面是Spark学习路线图：另外，提供小伙伴们几个spark学习资源：Spark亚太研究院王家林老师的spark公开课和收费视频，在2014年Spark亚太研究院流传的比较广泛，小伙伴们百度一下，就应该可以找到相关的QQ群、公开课和视频，其合作伙伴是51cto。Youtube上订阅Apache Spark和Databricks在路上的Scala学习资料还有几个meetup上可以获取最新的应用资料：http://www.meetup.com/spark-users/http://www.meetup.com/spark-user- ...继续阅读 (39)

经过一段时间授课，积累下不少的spark知识，想逐步汇总成资料，分享给小伙伴们。对于想视频学习的小伙伴，可以访问炼数成金网站的《spark大数据平台》课程，每周的课程是原理加实际操作。最新的课程是第3期，至于费用，越认真学习，学习成本越低。       从Spark1.1.0开始，重新规划了一下学习路线，由于涉及的内容比较多，会不断的完善其中的内容，争取在Spark1.2.0形成一个完整的体 ...继续阅读 (7)

spark是一个快速的内存计算框架；同时是一个并行运算的框架。在计算性能调优的时候，除了要考虑广为人知的木桶原理外，还要考虑平行运算的Amdahl定理。木桶原理又称短板理论，其核心思想是：一只木桶盛水的多少，并不取决于桶壁上最高的那块木块，而是取决于桶壁上最短的那块。将这个理论应用到系统性能优化上，系统的最终性能取决于系统中性能表现最差的组件。例如，即使系统拥有充足的内存资源和CPU资源，但是如果磁盘I/O性能低下，那么系统的总体性能是取决于当前最慢的磁盘I/O速度，而不是当前最优越的CPU或者内存。在这种情况下，如果需要进一步提升系统性能，优化内存或者CPU资源是毫无用处的。只有提高磁盘I/O性能才能对系统的整体性能进行优化。Amdahl定理，一个计算机科学界的经验法则，因吉恩·阿姆达尔而得名。它代表了处理器平行运算之后效率提升的能力。并行计算中的加速比是用并行前的执行速度和并行后的执行速度之比来表示的，它表示了在并行化之后的效率提升情况。阿姆达尔定律是固定负载（计算总量不变时）时的量化标准。可用公式：来表示。式中分别表示问题规模的串行分量（问题中不能并行化的那一部分）和并行分量，p表示处理器数量。当时，上式的极限是，其中，。这意味着无论我们如何增大处理器数目，加速比是无法高于这个数的。SparkSQL作为Spark的一个组件，在调优的时候，也要充分考虑到上面的两个原理，既要考虑 ...继续阅读 (78)

spark是一个快速的内存计算框架；同时是一个并行运算的框架。在计算性能调优的时候，除了要考虑广为人知的木桶原理外，还要考虑平行运算的Amdahl定理。       木桶原理又称短板理论，其核心思想是：一只木桶盛水的多少，并不取决于桶壁上最高的那块木块，而是取决于桶壁上最短的那块。将这个理论应用到系统性能优化上，系统的最终性能取决于系统中性能表现最差的组件。例如，即使系统拥有充足的内存资源和 ...继续阅读 (7)

回顾一下，在前面几章中，就sparkSQL1.1.0基本概念、运行架构、基本操作和实用工具做了基本介绍。基本概念：SchemaRDDRuleTreeLogicPlanParserAnalyzerOptimizerSparkPlan运行架构：sqlContext运行架构hiveContext运行架构基本操作原生RDD的操作parquet文件的操作json文件的操作hive数据的操作和其他spark组件混合使用实用工具hive/console的操作CLI的配置和操作ThriftServer的配置和操作由于时间仓促，有很多地方来不及详细，特别是第三章和第九章；另外还有一些新的特性没有介绍，比如列存储的实现过程、CODEGEN的源码分析等，将在后续的版本逐步完善。从总体上来说，由于CLI的引入，使得sparkSQL1.1.0在易用性方面得到了极大地提高；而ThriftServer的引入，方便了开发者对基于SparkSQL的应用程序开发；hive/console的引入，极大地方面了开发者对sparkSQL源码的修改和调试；还有json数据的引入，不但扩充了sparkSQL的数据来源，同时对嵌套数据开始做了尝试。从Spark1.1.0开始，sparkSQL逐渐开始像是一个产品了，而不像spark1.0.0，感觉像是一个测试品。当然，由于sparkSQL项目的启动时间比较晚，到现在为止还不到一年 ...继续阅读 (32)

回顾一下，在前面几章中，就sparkSQL1.1.0基本概念、运行架构、基本操作和实用工具做了基本介绍。 基本概念： SchemaRDD RuleTreeLogicPlanParserAnalyzerOptimizerSparkPlan 运行架构： sqlContext运行架构hiveContext运行架构 基本操作 原生RDD的操作parquet文件的操作j ...继续阅读 (5)

前面两章花了不少篇幅介绍了SparkSQL的运行过程，很多读者还是觉得其中的概念很抽象，比如Unresolved LogicPlan、LogicPlan、PhysicalPlan是长得什么样子，没点印象，只知道名词，感觉很缥缈。本章就着重介绍一个工具hive/console，来加深读者对sparkSQL的运行计划的理解。1：hive/console安装sparkSQL从1.0.0开始提供了一个sparkSQL的调试工具hive/console。该工具是给开发者使用，在编译生成的安装部署包中并没有；该工具需要使用sbt编译运行。要使用该工具，需要具备以下条件：spark1.1.0源码hive0.12源码并编译配置环境变量1.1：安装hive/cosole下面是笔者安装过程：A：下载spark1.1.0源码，安装在/app/hadoop/spark110_sql目录B：下载hive0.12源码，安装在/app/hadoop/hive012目录，进入src目录后，使用下面命令进行编译：ant clean package -Dhadoop.version=2.2.0 -Dhadoop-0.23.version=2.2.0 -Dhadoop.mr.rev=23C：配置环境变量文件～/.bashrc后，source ～/.bashrc使环境变量生效。export HIVE_HOME=/app/ ...继续阅读 (92)

前面两章花了不少篇幅介绍了SparkSQL的运行过程，很多读者还是觉得其中的概念很抽象，比如Unresolved LogicPlan、LogicPlan、PhysicalPlan是长得什么样子，没点印象，只知道名词，感觉很缥缈。本章就着重介绍一个工具hive/console，来加深读者对sparkSQL的运行计划的理解。 1：hive/console安装       spa ...继续阅读 (9)

在介绍sparkSQL之前，我们首先来看看，传统的关系型数据库是怎么运行的。当我们提交了一个很简单的查询：SELECT a1,a2,a3 FROM tableA Where condition可以看得出来，该语句是由Projection（a1，a2，a3）、Data Source（tableA）、Filter（condition）组成，分别对应sql查询过程中的Result、Data Source、Operation，也就是说SQL语句按Result-->Data Source-->Operation的次序来描述的。那么，SQL语句在实际的运行过程中是怎么处理的呢？一般的数据库系统先将读入的SQL语句（Query）先进行解析（Parse），分辨出SQL语句中哪些词是关键词（如SELECT、FROM、WHERE），哪些是表达式、哪些是Projection、哪些是Data Source等等。这一步就可以判断SQL语句是否规范，不规范就报错，规范就继续下一步过程绑定（Bind），这个过程将SQL语句和数据库的数据字典（列、表、视图等等）进行绑定，如果相关的Projection、Data Source等等都是存在的话，就表示这个SQL语句是可以执行的；而在执行前，一般的数据库会提供几个执行计划，这些计划一般都有运行统计数据，数据库会在这些计划中选择一个最优计划（Optimize ...继续阅读 (74)

在介绍sparkSQL之前，我们首先来看看，传统的关系型数据库是怎么运行的。当我们提交了一个很简单的查询： SELECT a1,a2,a3 FROM tableA Where condition 可以看得出来，该语句是由Projection（a1，a2，a3）、Data Source（tableA）、Filter（condition）组成，分别对应sql查询过程中的Result ...继续阅读 (5)

2014年9月11日，Spark1.1.0忽然之间发布。笔者立即下载、编译、部署了Spark1.1.0。关于Spark1.1的编译和部署，请参看笔者博客Spark1.1.0 源码编译和部署包生成。Spark1.1.0中变化较大是sparkSQL和MLlib，sparkSQL1.1.0主要的变动有：增加了JDBC/ODBC Server（ThriftServer），用户可以在应用程序中连接到SparkSQL并使用其中的表和缓存表。增加了对JSON文件的支持增加了对parquet文件的本地优化增加了支持将python、scala、java的lambda函数注册成UDF，并能在SQL中直接引用引入了动态字节码生成技术（bytecode generation，即CG），明显地提升了复杂表达式求值查询的速率。统一API接口，如sql()、SchemaRDD生成等。......下面分十个小节来介绍sparkSQL1.1.0的架构和使用，希望各位读者joy it！第一节：为什么sparkSQL 为本篇，介绍sparkSQL的发展历程和性能第二节：sparkSQL架构 介绍catalyst，然后介绍sqlContext、hiveContext的运行架构及区别第三节：sparkSQL组件之解析 介绍sparkSQL运行架构中的各个组件的功能和实现第四节：深入了解sparkS ...继续阅读 (247)

2014年9月11日，Spark1.1.0忽然之间发布。笔者立即下载、编译、部署了Spark1.1.0。关于Spark1.1的编译和部署，请参看笔者博客Spark1.1.0 源码编译和部署包生成 。       Spark1.1.0中变化较大是sparkSQL和MLlib，sparkSQL1.1.0主要的变动有： 增加了JDBC/ODBC Server（ThriftServer）， ...继续阅读 (5)

Spark之所以万人瞩目，除了内存计算，还有其ALL-IN-ONE的特性，实现了One stack rule them all。下面简单模拟了几个综合应用场景，不仅使用了sparkSQL，还使用了其他Spark组件：店铺分类，根据销售额对店铺分类，使用sparkSQL和MLLibPageRank，计算最有价值的网页，使用sparkSQL和GraphX前者将使用sparkSQL+MLlib的聚类算法，后者将使用sparkSQL+GraphX的PageRank算法。本实验采用IntelliJ IDEA调试代码，最后生成doc.jar，然后使用spark-submit提交给集群运行。1：店铺分类分类在实际应用中非常普遍，比如对客户进行分类、对店铺进行分类等等，对不同类别采取不同的策略，可以有效的降低企业的营运成本、增加收入。机器学习中的聚类就是一种根据不同的特征数据，结合用户指定的类别数量，将数据分成几个类的方法。下面举个简单的例子，对第五小结中的hive数据，按照销售数量和销售金额这两个特征数据，进行聚类，分出3个等级的店铺。在IDEA中建立一个object：SQLMLlibpackage doc import org.apache.log4j.{Level, Logger} import org.apache.spark.sql.catalyst.ex ...继续阅读 (61)

Spark之所以万人瞩目，除了内存计算，还有其ALL-IN-ONE的特性，实现了One stack rule them all。下面简单模拟了几个综合应用场景，不仅使用了sparkSQL，还使用了其他Spark组件： 店铺分类，根据销售额对店铺分类货品调拨，根据货品的销售数量和店铺之间的距离进行货品调拨       前者将使用sparkSQL+MLlib的聚类算法，后者将使用s ...继续阅读 (5)

SparkSQL引入了一种新的RDD——SchemaRDD，SchemaRDD由行对象（row）以及描述行对象中每列数据类型的schema组成；SchemaRDD很象传统数据库中的表。SchemaRDD可以通过RDD、Parquet文件、JSON文件、或者通过使用hiveql查询hive数据来建立。SchemaRDD除了可以和RDD一样操作外，还可以通过registerTempTable注册成临时表，然后通过SQL语句进行操作。值得注意的是：Spark1.1使用registerTempTable代替1.0版本的registerAsTableSpark1.1在hiveContext中，hql()将被弃用，sql()将代替hql()来提交查询语句，统一了接口。使用registerTempTable注册表是一个临时表，生命周期只在所定义的sqlContext或hiveContext实例之中。换而言之，在一个sqlontext（或hiveContext）中registerTempTable的表不能在另一个sqlContext（或hiveContext）中使用。另外，spark1.1提供了语法解析器选项spark.sql.dialect，就目前而言，spark1.1提供了两种语法解析器：sql语法解析器和hiveql语法解析器。sqlContext现在只支持sql语法解析器（S ...继续阅读 (92)

sparkSQL1.1对数据的查询分成了2个分支：sqlContext 和 hiveContext。       在sqlContext中，sparkSQL可以使用SQL-92语法对定义的表进行查询，表的源数据可以来自： RDDparquet文件json文件       在hiveContext中，sparkSQL可以使用HQL语法，对hive数据进行查询，sparkSQ ...继续阅读 (7)

前面介绍了sparkSQL的运行架构，后面将介绍sparkSQL的使用。在介绍sparkSQL的使用之前，我们需要搭建一个sparkSQL的测试环境。本次测试环境涉及到hadoop之HDFS、hive、spark以及相关的数据文件，相关的信息如下：hadoop版本为2.2.0hive版本为0.13spark版本为1.1.0MySQL版本为5.6.12测试数据下载地点：http://pan.baidu.com/s/1eQCbT30#path=%252Fblog中的sparkSQL_data.zip测试环境示意图：本测试环境是在一台物理机上搭建的，物理机的配置是16G内存，4核8线程CPU。hadoop1、hadoop2、hadoop3是vitual box虚拟机，构建hadoop集群和spark集群；物理机wyy作为客户端，编写代码和提交计算任务。总的测试环境配置如下：机器名配置角色软件安装hadoop14G内存，1核hadoop：NN/DN Spark：Master/worker/app/hadoop/hadoop220/app/hadoop/spark110/app/scala2104/usr/java/jdk1.7.0_21hadoop24G内存，1核hadoop：DN Spark：workerhive0.13客户端/app/hadoop/hadoop220/app/hadoo ...继续阅读 (38)

前面介绍了sparkSQL的运行架构，后面将介绍sparkSQL的使用。在介绍sparkSQL的使用之前，我们需要搭建一个sparkSQL的测试环境。本次测试环境涉及到hadoop之HDFS、hive、spark以及相关的数据文件，相关的信息如下： hadoop版本为2.2.0hive版本为0.13spark版本为1.1.0-rc3MySQL版本为5.6.12测试数据下载地点：http: ...继续阅读 (4)

spark1.1相较于spark1.0，最大的差别就在于spark1.1增加了万人期待的CLI和ThriftServer。使得hive用户还有用惯了命令行的RDBMS数据库管理员很容易地上手sparkSQL，在真正意义上进入了SQL时代。下面先简单介绍其使用，限于时间关系，以后再附上源码分析。1：令人惊讶的CLI刚部署好spark1.1就迫不及待地先测试CLI（bin/spark-sql），对于习惯了sql命令行的本人，失去了shark后，对于sparkSQL1.0一度很是抵触（其实对于开发调试人员来说，spark-shell才是利器，可以很方便地使用各个spark生态中的组件）。急切中，没有关闭hive metastore服务，然后一个bin/spark-sql就进入了命令行，然后通过hive metastore就可以直接对hive进行查询了：spark-sql> use saledata; //所有订单中每年的销售单数、销售总额 spark-sql> select c.theyear,count(distinct a.ordernumber),sum(b.amount) from tblStock a join tblStockDetail b on a.ordernumber=b.ordernumber join tbldate c on a.dateid=c.dateid ...继续阅读 (147)

spark1.1相较于spark1.0，最大的差别就在于spark1.1增加了万人期待的CLI和ThriftServer。使得hive用户还有用惯了命令行的RDBMS数据库管理员很容易地上手sparkSQL，在真正意义上进入了SQL时代。下面先简单介绍其使用，限于时间关系，以后再附上源码分析。 1：ThriftServer和CLI的命令参数 A：令人惊讶的CLI ...继续阅读 (3)

今天准备将mysql的数据倒腾到RDD，很早以前就知道有一个JdbcRDD，就想着使用一下，结果发现却是鸡肋一个。首先，看看JdbcRDD的定义：* An RDD that executes an SQL query on a JDBC connection and reads results. * For usage example, see test case JdbcRDDSuite. * * @param getConnection a function that returns an open Connection. * The RDD takes care of closing the connection. * @param sql the text of the query. * The query must contain two ? placeholders for parameters used to partition the results. * E.g. "select title, author from books where ? <= id and id <= ?" * @param lowerBound the minimum value of the first placeholder * @param upper ...继续阅读 (44)

今天准备将mysql的数据倒腾到RDD，很早以前就知道有一个JdbcRDD，就想着使用一下，结果发现却是鸡肋一个。       首先，看看JdbcRDD的定义： * An RDD that executes an SQL query on a JDBC connection and reads results. * For usage example, see test case ...继续阅读 (11)

Spark Application可以在集群中并行运行，其关键是抽象出RDD的概念（详见RDD 细解），也使得Spark Application的开发变得简单明了。下图浓缩了Spark的编程模型。1：Spark应用程序的结构Spark应用程序可分两部分：driver部分和executor部分初始化SparkContext和主体程序A：driver部分driver部分主要是对SparkContext进行配置、初始化以及关闭。初始化SparkContext是为了构建Spark应用程序的运行环境，在初始化SparkContext，要先导入一些Spark的类和隐式转换；在executor部分运行完毕后，需要将SparkContext关闭。driver部分的基本代码框架如下：package week2 import org.apache.spark.{SparkContext, SparkConf} import org.apache.spark.SparkContext._ object WordCount1 { def main(args: Array[String]) { if (args.length == 0) { System.err.println("Usage: bin/spark-submit [options] --class week2.W ...继续阅读 (35)

Spark Application可以在集群中并行运行，其关键是抽象出RDD的概念（详见RDD 细解），也使得Spark Application的开发变得简单明了。下图浓缩了Spark的编程模型。       1：Spark应用程序的结构       Spark应用程序可分两部分：driver部分和executor部分初始化SparkContext和主体程序 ...继续阅读 (6)

在运行Spark应用程序的时候，driver会提供一个webUI给出应用程序的运行信息，但是该webUI随着应用程序的完成而关闭端口，也就是说，Spark应用程序运行完后，将无法查看应用程序的历史记录。Spark history server就是为了应对这种情况而产生的，通过配置，Spark应用程序在运行完应用程序之后，将应用程序的运行信息写入指定目录，而Spark history server可以将这些运行信息装载并以web的方式供用户浏览。要使用history server，对于提交应用程序的客户端需要配置以下参数（在conf/spark-defaults.conf中配置）：spark.eventLog.enabled是否记录Spark事件，用于应用程序在完成后重构webUI。spark.eventLog.dir如果spark.eventLog.enabled为 true，该属性为记录spark事件的根目录。在此根目录中，Spark为每个应用程序创建分目录，并将应用程序的事件记录到在此目录中。用户可以将此属性设置为HDFS目录，以便history server读取历史记录文件。spark.yarn.historyServer.addressSpark history server的地址（不要加http://）。这个地址会在Spark应用程序完成后提交给YARN RM，然后RM ...继续阅读 (54)

在运行Spark应用程序的时候，driver会提供一个webUI给出应用程序的运行信息，但是该webUI随着应用程序的完成而关闭端口，也就是说，Spark应用程序运行完后，将无法查看应用程序的历史记录。Spark history server就是为了应对这种情况而产生的，通过配置，Spark应用程序在运行完应用程序之后，将应用程序的运行信息写入指定目录，而Spark history serve ...继续阅读 (8)

Spark1.0.0可以通过以下几种方式来对Spark应用程序进行监控：Spark应用程序的WebUI或者Spark Standalone的集群监控指标，然后通过支持指标收集的集群监控系统，如ganglia进行监控辅助监控工具1：WebUISpark应用程序提交后，driver和Executor之间不断的交换运行信息，可以通过driver的4040端口（默认端口）获取有用的Spark应用程序的运行信息，如：Stage和TaskRDD大小和内存使用情况环境变量信息executor的运行信息...如果多个Spark应用程序在同一个client上以client方式提交，那么driver的WebUI端口将绑定从4040开始的连续端口，如4040、4041、4042...。需要注意的是，用过WebUI只能查看Spark应用程序在运行期间的信息，一旦Spark应用程序运行完，这些信息将无法查看，因为WebUI端口随Spark应用程序的完成而关闭。如果想要事后查看Spark应用程序的运行信息，那么需要配置history Server来持久化Spark应用程序运行信息。关于history Server参见Spark1.0.0 history server配置（正在撰写，迟点给上链接） 。2：指标Spark采用了基于Coda Hale Metrics Library的可配置的指标体系，通过各种指标收 ...继续阅读 (54)

Spark1.0.0可以通过以下几种方式来对Spark应用程序进行监控： Spark应用程序的WebUI或者Spark Standalone的集群监控指标，然后通过支持指标收集的集群监控系统，如ganglia进行监控辅助监控工具 1：WebUI       Spark应用程序提交后，driver和Executor之间不断的交换运行信息，可以通过driver的4 ...继续阅读 (10)