ETL是 Extract-Transform-Load的缩写，也就是抽取-转换-加载，在数据工作中是非常重要的部分。实际上，ETL就是一个对数据进行批处理的过程，一个ETL程序就是一个批处理脚本，执行时能将一堆数据转化成我们需要的形式。 每个接触过数据批处理的工程师，都走过ETL的流程，只是没有意识到而已。按照ETL过程的框架来重新认识数据批处理，有利于我们更清晰地编写批处理脚本。 在单机范围内的数据量下，使用python的pandas包就可以非常方便地完成数据批处理工作。但当数据量达到1G以上时，pandas处理起来就有些力不从心了，到数据量达到1T以上，只能以分块的方式存储在分布式系统上时，pandas就无能为力了。在当前的技术背景下，典型的场景就是数据存储在Hive on HDFS上。要做ETL，就需要新的工具。Hadoop生态下，原生的工具是MapReduce计算模型，通常用Java编写，比较复杂，每次计算的中间结果也需要进行磁盘存取，非常费时。Spark是一个MPP架构的计算引擎，相比MapReduce，Spark 有DataFrame（又名 Schema RDD）, 以表的形式来储存数据，无论是理解还是操作，都更为简单，还支持Python，在许多需要使用函数作参数的场合，非常好用。

本教程将介绍如何使用pyspark.sql模块，操作Spark DataFrame，从Hive中读取数据，经过一系列转换，最后存入Hive中。Spark的DataFrame和pandas的DataFrame的概念很相似，只是操作略有不同，如果读者有pandas的使用经验，很容易就能快速上手。 教程只是为了方便读者快速入门，想要更好地开发Spark程序，仍然需要详细了解Spark的API接口，对python环境下，Hive的ETL来说，研究pyspark.sql模块下的内容就足够了，可以参考官方文档。

环境：Spark的API随版本不同会有较大变化，目前比较流行的版本是1.6和2.2，本文使用Spark 1.6.0，语言为Python 2.7。默认数据都储存在Hive中，Hadoop集群带有yarn。

学习一门语言或者软件的第一步，永远都是冒烟测试。最经典的冒烟测试就是输出Hello World。但对ETL来说，一个打印"Hello World"的Spark程序是没什么用的。所以我们这里讲讲如何打印一张表，这张表有一行数据，列名为t,值为"Hello World"。

Spark的核心是SparkContext，它提供了Spark程序的运行环境。而SqlContext则是由SparkContext产生，提供了对数据库表的访问接口。因为这里数据库的环境是Hive，通常使用SqlContext的派生类HiveContext。在Spark提供的交互式环境中，会在启动时自动创建环境，生成SparkContext和HiveContext的实例。在pyspark的交互式环境中，SparkContext实例名为sc，HiveContext实例名为sqlContext。

交互式操作只在学习和调试时使用，实际工作中还是要靠命令行执行脚本。在脚本中我们就需要自己生成SparkContext和HiveContext了。基本操作代码如下：

将这段代码保存成文件hello.py，在终端中进入到该文件所在目录，输入命令spark-submit --master yarn hello.py ，然后就可以看到屏幕上输出如下，冒烟测试就算完成了。

指令解释：spark-submit就是spark的执行程序，master yarn是spark-submit的参数，指定yarn作为计算调度的中心。最后hello.py就是我们的ETL程序。

ETL的第一步就是从数据源抽取数据，在Spark中就是从Hive里读取数据。

Hive虽然实质上是个MapReduce接口的封装，但从上层抽象模型来看，有最基本的Schema、Table和Column，还有一套类SQL语法，可以说就是一个典型的关系数据库模型，因此在ETL过程中，我们完全可以把Hive当成一个关系数据库来看待。

抽取的常用方法由两种，一种是直接读取Hive表，一种是通过Hive QL读取。 都需要以HiveContext的实例作为入口，结果返回一个Spark DataFrame，为了检查结果，可以使用show方法查看DataFrame的数据。

假设我们有一个名为test 的库，里面有一张表为t1，数据结构如下：

HiveContext对读取操作提供统一的接口- DataFrameReader，HiveContext的实例的read属性就可以获取到这个接口。 当然，这个接口也能用来读取Hive的数据，read.table就可获取到表的数据，代码如下

实质是让HiveContext将HiveQL传给Hive,让Hive执行后，将查询结果封装成Spark DataFrame返回。在处理过程比较简单，或者需要大量设置别名时，比较有用（因为Spark批量设置别名不太方便），但不推荐写太过复杂的Hive QL，因为Hive 执行Hive QL的实质是把Hive QL转成MapReduce执行，在计算效率上是不如Spark的。

为什么不先讲Trasform呢？因为Trasform的操作很多，先讲Load有助于快速上手完成一个初级的ETL程序。 类似于读取，HiveContext也提供了统一的写接口，名为DataFrameWriter.调用write属性即可获取。

写入的具体方式也很多，不过为了快速上手，只讲最关键的一些东西。

如果表已经存在，该如何操作。

因为Hive on HDFS的关系，更新表最快的方式是全表覆写。对于需要更新原有的ETL，一般都是全表重写，只需要追加的，就可以用追加。

在Hive on HDFS中，数据实质上是以文件的形式保存的。不同的文件格式，在压缩容量、支持数据类型和查询速度上都有所不同。textfile,avro,sequence,parquet,json等。目前我常用的格式是text和parquet，如果不设置文件格式，默认会使用Hive表的文件格式，如果Hive表不存在，则使用Hive表的默认格式textfile

了解了上面的操作之后，我们就可以开始写加载部分的代码了，只需要使用一个saveAsTable方法就行了，非常简单。

转换是ETL过程中最复杂的部分，去掉抽取和加载，剩下的全都是转换，包含的内容是非常多的，常见的有筛选、聚合、多列合并或计算，列赋值，根据不同的需要有不同的处理方法。由于Spark的转换操作较为啰嗦，所以推荐把部分简单的操作通过Hive QL的方式，在抽取步骤中交由Hive完成，这样有助于精简代码，提高可读性，降低维度难度。 下面就讲一讲Spark DataFrame 转换部分的基本概念和操作。

对于日常用Java来做数据批处理的工程师来说，可能更习惯用for循环来逐条处理数据。但这样做在操作上是很不方便的，也不太利于阅读理解。在科学计算的语境下，数据总是以DataFrame的形式储存，也就是一张表。数据处理操作通常是对这张表的某些行或者某些列来进行处理。比如，“令t1表的a列中数字大于2的值的，全部都等于2”，或者“给t1表新加一常数列d，值为99”，这样的操作在向量化编程的语境下，就是一个调用API接口的操作，比for循环容易被理解。 可以类比pandas。在pandas中，也主要是通过向量化编程的方式来处理数据，虽然提供了迭代器的接口，可以一行行地读取数据，但一般以表作为修改对象的操作，主要是以API接口来完成，不推荐使用迭代器来做行级修改。一来操作不方便，二来运算速度未必能比优化过的API接口快。 Spark是分布式执行的，数据分散在各个机器上，背后有一套调度系统来控制数据计算负载。如果用for循环来处理，就是把负载都加在了执行脚本的机器上，一般来说执行脚本的机器都是不储存数据的master，实际上这一过程就会导致需要把数据从slave传到master上，无谓地增加了网络负担。所以，在Spark脚本里，严禁使用原生的python for循环来处理SparkData Frame，即使要用，也应该使用Spark提供的API接口。

在Spark DataFrame语境下，操作对象主要有三个：DataFrame,Row,Column。

事实上，最常用的主要是DataFrame和Column，Row很少用到。其中，DataFrame是核心，一个ETl过程，实质就是从抽取一个DataFrame开始，经过一系列的DataFrame变换，得到一个与目标一致的DataFrame，然后写入到目标数据库中去。Column在其中扮演着中间点的角色，比如取DataFrame的多个列，拼接合成一个新列，然后把这个新列加到原本的DataFrame中去。

上面提到了，DataFrame是核心操作对象。其实在Spark中，真正意义上的核心操作对象是RDD，一个有序的，分布式储存在内存中的操作对象。DataFrame就是一个特殊的RDD——Schema RDD。所有的DataFrame操作，都可以归类为两种基本操作：转化（Transformation）和行动（action）。转换操作是不会触发Spark的实际计算的，即使转换过程中出现了错误，在执行到这一行代码时，也不会报错。直到执行了行动操作之后，才会真正让Spark执行计算，这时候才会抛出在转化过程中出现的错误。这在DEBU时，尤其是交互式编程环境下，可能会导致问题代码定位错误，需要特别注意。

filter(cond):筛选出满足条件cond的行。cond可以填字符串，格式和SQL中的where子句一样，也可以填Bool类型的Column对象，比如 df['a']>1。