Spark SQL中的DataFrame类似于一张关系型数据表。在关系型数据库中对单表或进行的查询操作，在DataFrame中都可以通过调用其API接口来实现。可以参考，Scala提供的 DataFrame API 。

本文中的代码基于Spark-1.6.2的文档实现。

一、DataFrame对象的生成

Spark-SQL可以以其他RDD对象、parquet文件、json文件、hive表，以及通过JDBC连接到其他关系型数据库作为数据源来生成DataFrame对象。本文将以MySQL数据库为数据源，生成DataFrame对象后进行相关的DataFame之上的操作。
文中生成DataFrame的代码如下：

object DataFrameOperations {
  def main (args: Array[String ]) {
    val sparkConf = new SparkConf().setAppName( "Spark SQL DataFrame Operations").setMaster( "local[2]" )
    val sparkContext = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sparkContext)
    val url = "jdbc:mysql://m000:3306/test"
    val jdbcDF = sqlContext.read.format( "jdbc" ).options(
      Map( "url" -> url,
        "user" -> "root",
        "password" -> "root",
        "dbtable" -> "spark_sql_test" )).load()
    val joinDF1 = sqlContext.read.format( "jdbc" ).options(
      Map("url" -> url ,
        "user" -> "root",
        "password" -> "root",
        "dbtable" -> "spark_sql_join1" )).load()
    val joinDF2 = sqlContext.read.format( "jdbc" ).options(
      Map ( "url" -> url ,
        "user" -> "root",
        "password" -> "root",
        "dbtable" -> "spark_sql_join2" )).load()
    ... ...
　　后续代码都在上面... ...处。
 
二、DataFrame对象上Action操作
 
1、show：展示数据
 
　　以表格的形式在输出中展示jdbcDF中的数据，类似于select * from spark_sql_test的功能。 
 　　show方法有四种调用方式，分别为， 
 （1）show 
 　　只显示前20条记录。 
 　　示例：
 
jdbcDF.show
 
（2）show(numRows: Int) 
 　　显示numRows条 
 　　示例：
 
jdbcDF.show(3)
 
（3）show(truncate: Boolean) 
 　　是否最多只显示20个字符，默认为true。 
 　　示例：
 
jdbcDF.show(true)
jdbcDF.show(false)
 
（4）show(numRows: Int, truncate: Boolean) 
 　　综合前面的显示记录条数，以及对过长字符串的显示格式。 
 　　示例：
 
jdbcDF.show(3, false)
 
2、collect：获取所有数据到数组
 
　　不同于前面的show方法，这里的collect方法会将jdbcDF中的所有数据都获取到，并返回一个Array对象。
 
jdbcDF.collect()
 
　　结果如下，结果数组包含了jdbcDF的每一条记录，每一条记录由一个GenericRowWithSchema对象来表示，可以存储字段名及字段值。 
 　　
 
3、collectAsList：获取所有数据到List
 
　　功能和collect类似，只不过将返回结构变成了List对象，使用方法如下
 
jdbcDF.collectAsList()
 
　　结果如下， 
 　　
 
4、describe(cols: String*)：获取指定字段的统计信息
 
　　这个方法可以动态的传入一个或多个String类型的字段名，结果仍然为DataFrame对象，用于统计数值类型字段的统计值，比如count, mean, stddev, min, max等。 
 　　使用方法如下，其中c1字段为字符类型，c2字段为整型，c4字段为浮点型
 
jdbcDF .describe("c1" , "c2", "c4" ).show()
 
　　结果如下， 
 　　
 
5、first, head, take, takeAsList：获取若干行记录
 
　　这里列出的四个方法比较类似，其中 
 　　（1）first获取第一行记录 
 　　（2）head获取第一行记录，head(n: Int)获取前n行记录 
 　　（3）take(n: Int)获取前n行数据 
 　　（4）takeAsList(n: Int)获取前n行数据，并以List的形式展现 
 　　以Row或者Array[Row]的形式返回一行或多行数据。first和head功能相同。 
 　　take和takeAsList方法会将获得到的数据返回到Driver端，所以，使用这两个方法时需要注意数据量，以免Driver发生OutOfMemoryError
 
　　使用和结果略。
 
二、DataFrame对象上的条件查询和join等操作
 
　　以下返回为DataFrame类型的方法，可以连续调用。
 
1、where条件相关
 
（1）where(conditionExpr: String)：SQL语言中where关键字后的条件 
 　　传入筛选条件表达式，可以用and和or。得到DataFrame类型的返回结果， 
 　　示例：
 
jdbcDF .where("id = 1 or c1 = 'b'" ).show()
 
（2）filter：根据字段进行筛选 
 　　传入筛选条件表达式，得到DataFrame类型的返回结果。和where使用条件相同 
 　　示例：
 
jdbcDF .filter("id = 1 or c1 = 'b'" ).show()
 
2、查询指定字段
 
（1）select：获取指定字段值 
 　　根据传入的String类型字段名，获取指定字段的值，以DataFrame类型返回 
 　　示例：
 
jdbcDF.select( "id" , "c3" ).show( false)
 
　　还有一个重载的select方法，不是传入String类型参数，而是传入Column类型参数。可以实现select id, id+1 from test这种逻辑。
 
jdbcDF.select(jdbcDF( "id" ), jdbcDF( "id") + 1 ).show( false)
 
　　能得到Column类型的方法是apply以及col方法，一般用apply方法更简便。
 
（2）selectExpr：可以对指定字段进行特殊处理 
 　　可以直接对指定字段调用UDF函数，或者指定别名等。传入String类型参数，得到DataFrame对象。 
 　　示例，查询id字段，c3字段取别名time，c4字段四舍五入：
 
jdbcDF .selectExpr("id" , "c3 as time" , "round(c4)" ).show(false)
 
（3）col：获取指定字段 
 　　只能获取一个字段，返回对象为Column类型。 
 　　val idCol = jdbcDF.col(“id”)果略。
 
（4）apply：获取指定字段 
 　　只能获取一个字段，返回对象为Column类型 
 　　示例：
 
val idCol1 = jdbcDF.apply("id")
val idCol2 = jdbcDF("id")
 
（5）drop：去除指定字段，保留其他字段 
 　　返回一个新的DataFrame对象，其中不包含去除的字段，一次只能去除一个字段。 
 　　示例：
 
jdbcDF.drop("id")
jdbcDF.drop(jdbcDF("id"))
 
3、limit
 
　　limit方法获取指定DataFrame的前n行记录，得到一个新的DataFrame对象。和take与head不同的是，limit方法不是Action操作。
 
jdbcDF.limit(3).show( false)
 
4、order by
 
（1）orderBy和sort：按指定字段排序，默认为升序 
 　　示例1，按指定字段排序。加个-表示降序排序。sort和orderBy使用方法相同
 
jdbcDF.orderBy(- jdbcDF("c4")).show(false)
// 或者
jdbcDF.orderBy(jdbcDF("c4").desc).show(false)
 
　　示例2，按字段字符串升序排序
 
jdbcDF.orderBy("c4").show(false)
 
（2）sortWithinPartitions 
 　　和上面的sort方法功能类似，区别在于sortWithinPartitions方法返回的是按Partition排好序的DataFrame对象。
 
5、group by
 
（1）groupBy：根据字段进行group by操作 
 　　groupBy方法有两种调用方式，可以传入String类型的字段名，也可传入Column类型的对象。 
 　　使用方法如下，
 
jdbcDF .groupBy("c1" )
jdbcDF.groupBy( jdbcDF( "c1"))
 
（2）cube和rollup：group by的扩展 
 　　功能类似于SQL中的group by cube/rollup，略。
 
（3）GroupedData对象 
 　　该方法得到的是GroupedData类型对象，在GroupedData的API中提供了group by之后的操作，比如，
 
max(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的最大值，只能作用于数字型字段
min(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的最小值，只能作用于数字型字段
mean(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的平均值，只能作用于数字型字段
sum(colNames: String*)方法，获取分组中指定字段或者所有的数字类型字段的和值，只能作用于数字型字段
count()方法，获取分组中的元素个数
 
　　运行结果示例： 
 　　count 
 　　
 
　　max 
 　　
 
　　这里面比较复杂的是以下两个方法， 
 agg，该方法和下面介绍的类似，可以用于对指定字段进行聚合操作。
 
pivot
 
6、distinct
 
（1）distinct：返回一个不包含重复记录的DataFrame 
 　　返回当前DataFrame中不重复的Row记录。该方法和接下来的dropDuplicates()方法不传入指定字段时的结果相同。 
 　　示例：
 
jdbcDF.distinct()
 
（2）dropDuplicates：根据指定字段去重 
 　　根据指定字段去重。类似于select distinct a, b操作 
 　　示例：
 
jdbcDF.dropDuplicates(Seq("c1"))
 
　　聚合操作调用的是agg方法，该方法有多种调用方式。一般与groupBy方法配合使用。 
 　　以下示例其中最简单直观的一种用法，对id字段求最大值，对c4字段求和。
 
jdbcDF.agg("id" -> "max", "c4" -> "sum")
 
8、union
 
　　unionAll方法：对两个DataFrame进行组合 
 　　类似于SQL中的UNION ALL操作。 
 　　示例：
 
jdbcDF.unionALL(jdbcDF.limit(1))
 
9、join
 
　　重点来了。在SQL语言中用得很多的就是join操作，DataFrame中同样也提供了join的功能。 
 　　接下来隆重介绍join方法。在DataFrame中提供了六个重载的join方法。 
 （1）、笛卡尔积
 
joinDF1.join(joinDF2)
 
（2）、using一个字段形式 
 　　下面这种join类似于a join b using column1的形式，需要两个DataFrame中有相同的一个列名，
 
joinDF1.join(joinDF2, "id")
 
　　joinDF1和joinDF2根据字段id进行join操作，结果如下，using字段只显示一次。 
 　　
 
（3）、using多个字段形式 
 　　除了上面这种using一个字段的情况外，还可以using多个字段，如下
 
joinDF1.join(joinDF2, Seq("id", "name")）
 
（4）、指定join类型 
 　　两个DataFrame的join操作有inner, outer, left_outer, right_outer, leftsemi类型。在上面的using多个字段的join情况下，可以写第三个String类型参数，指定join的类型，如下所示
 
joinDF1.join(joinDF2, Seq("id", "name"), "inner"）
 
（5）、使用Column类型来join 
 　　如果不用using模式，灵活指定join字段的话，可以使用如下形式
 
joinDF1.join(joinDF2 , joinDF1("id" ) === joinDF2( "t1_id"))
 
　　结果如下， 
 　　
 
（6）、在指定join字段同时指定join类型 
 　　如下所示
 
joinDF1.join(joinDF2 , joinDF1("id" ) === joinDF2( "t1_id"), "inner")
 
10、获取指定字段统计信息
 
　　stat方法可以用于计算指定字段或指定字段之间的统计信息，比如方差，协方差等。这个方法返回一个DataFramesStatFunctions类型对象。 
 　　下面代码演示根据c4字段，统计该字段值出现频率在30%以上的内容。在jdbcDF中字段c1的内容为"a, b, a, c, d, b"。其中a和b出现的频率为2 / 6，大于0.3
 
jdbcDF.stat.freqItems(Seq ("c1") , 0.3).show()
 
　　结果如下： 
 　　
 
11、获取两个DataFrame中共有的记录
 
　　intersect方法可以计算出两个DataFrame中相同的记录，
 
jdbcDF.intersect(jdbcDF.limit(1)).show(false)
 
　　结果如下： 
 　　
 
12、获取一个DataFrame中有另一个DataFrame中没有的记录
 
jdbcDF.except(jdbcDF.limit(1)).show(false)
 
　　结果如下， 
 　　
 
13、操作字段名
 
（1）withColumnRenamed：重命名DataFrame中的指定字段名 
 　　如果指定的字段名不存在，不进行任何操作。下面示例中将jdbcDF中的id字段重命名为idx。
 
jdbcDF.withColumnRenamed( "id" , "idx" )
 
　　结果如下： 
 　　
 
（2）withColumn：往当前DataFrame中新增一列 
 　　whtiColumn(colName: String , col: Column)方法根据指定colName往DataFrame中新增一列，如果colName已存在，则会覆盖当前列。 
 　　以下代码往jdbcDF中新增一个名为id2的列，
 
jdbcDF.withColumn("id2", jdbcDF("id")).show( false)
 
　　结果如下， 
 　　
 
14、行转列
 
　　有时候需要根据某个字段内容进行分割，然后生成多行，这时可以使用explode方法 
 　　下面代码中，根据c3字段中的空格将字段内容进行分割，分割的内容存储在新的字段c3_中，如下所示
 
jdbcDF.explode( "c3" , "c3_" ){time: String => time.split( " " )}
 
　　结果如下， 
 　　
 
15、其他操作
 
　　API中还有na, randomSplit, repartition, alias, as方法，待后续补充。
                    Spark SQL中的DataFrame类似于一张关系型数据表。在关系型数据库中对单表或进行的查询操作，在DataFrame中都可以通过调用其API接口来实现。可以参考，Scala提供的DataFrame API。　　本文中的代码基于Spark-1.6.2的文档实现。一、DataFrame对象的生成　　Spark-SQL可以以其他RDD对象、parquet文件、json文件、hive表，以及通过JD
Spark学习笔记（一）：Spark概述与运行原理 
Spark学习笔记（二）：RDD编程基础 
Spark SQL增加了DataFrame（即带有Schema信息的RDD），使用户可以在Spark SQL中执行SQL语句，数据既可以来自RDD，也可以是Hive、HDFS、Cassandra等外部数据源，还可以是JSON格式的数据
	Spark SQL目前支持Scala、Java、Python三种语言，支持SQL-92规范
	•DataFrame的推出，让Spark具备了处理大规模结构化数据的能力，不仅比原有的
				
Spark-Redis
 一个使用在读写数据的库。
 Spark-Redis可从Spark作为RDD访问Redis的所有数据结构-字符串，哈希，列表，集合和排序集合。 它还支持使用DataFrames和Spark SQL语法进行读写。
 该库可以与Redis独立数据库以及集群数据库一起使用。 与Redis群集一起使用时，Spark-Redis会意识到其分区方案，并会根据重新分片和节点故障事件进行调整。
 Spark-Redis还支持Spark流（DStream）和结构化流。
版本兼容性和分支
该库具有多个分支，每个分支对应于一个受支持的不同Spark版本。 例如，“ branch-2.3”可与
				
讲习班实践
在本研讨会中，练习集中于使用和 API，以及数据处理中的 。 我的github帐户（此处为scala）中的和Scala均提供了练习。 您只需要克隆项目就可以了！ 如果需要帮助，请查看解决方案分支。
 原始博客文章就。
 为了帮助您实现每节课，我们引入了单元测试。
 使用的框架：
 星火1.4.0
 Scala2.10
所有练习均作为独立程序在本地模式下运行。
 要动手操作，请通过以下命令行检索代码： 
 $ git clone https://github.com/nivdul/spark-in-practice-scala.git
 然后，您可以在IntelliJ或Eclipse中导入项目（添加用于Scala的SBT和Scala插件）​​，或使用崇高的文本。
 如果要使用交互式spark-shell（仅限scala / python），则需要下载。 
				
pySpark-flatten-dataframe
 PySpark函数可展平从JSON / CSV / SQL / Parquet加载的任何复杂的嵌套数据框结构
例如，对于嵌套的JSON-
 展平所有嵌套项：{“ human”：{“ name”：{“ first_name”：“ Jay Lohokare”}}}
 通过column ='human-name-first_name'转换为dataFrame。可以通过更改连接器变量来更改连接器'-'。
 爆炸数组：{“ array”：[“ one”，“ two”，“ three”]}转换为具有3行的column ='array'的dataFrame
 该函数可以处理任何级别的嵌套。
 该函数不能处理数组中的数组。 这只是为了保持代码的动态性和通用性。 为了处理内部数组数组，修改if isinstance在for的循环flattenSchema
				
一、定义数据框DataFrame
import  pandas
frame = pandas.DataFrame({"a":[9,2,5,1],"b":[4,7,-3,2],"c":[6,5,8,3]})
frame
Out[53]: 
   a  b  c
0  9  4  6
1  2  7  5
2  5 -3  8
3  1  2  3
二、按列对DataFrame排序
Action 操作
1、 collect() ,返回值是一个数组，返回dataframe集合所有的行
2、 collectAsList() 返回值是一个java类型的数组，返回dataframe集合所有的行
3、 count() 返回一个number类型的，返回dataframe集合的行数
4、 describe(cols: String*) 返回一个通过数学计算的类表值(count, mean, stddev, min, and max)，这个可以传多个参数，中间用逗号分隔，如
				
把以下sql语句，转为spark编程中DataFrame的同等形式：
select *, 
  row_number() over(partition by name order by id desc) rn 
from test_tb;
以下是等效内容：
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.desc
import org.apache.spark.sql.function
object DfSortDesc {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("DfSortDesc").master("local").getOrCreate()
    val data = Array((7,.
				
最好可以用RDD的就不要用DataFrame
今日就遇到执行出现 SparkContext异常停止，怀疑是DataFrame的distinct操作和groupby一样并不在本地合并为最小集，导致最后崩溃；而后换成RDD.distinct()却是可以的。
经多次测试都是以上结论
测试数据 一亿两千万条
结论：能用RDD的相关操作，就别用DataFrame，比如排序、统计count、disti...
				
Spark SQL是Apache Spark的一个模块，它提供了一种基于结构化数据的编程接口。Spark SQL支持使用SQL语言进行数据查询和操作，同时也支持使用DataFrame API进行编程。
Spark SQL的编程模型基于RDD（弹性分布式数据集）模型，但是它将RDD的抽象概念扩展到了结构化数据上。Spark SQL中的数据被组织成了一组命名的列，这些列可以被看作是表中的字段。Spark SQL还支持读取和写入多种数据源，包括Hive、JSON、Parquet等。
在Spark SQL中，用户可以使用SQL语言进行数据查询和操作，也可以使用DataFrame API进行编程。DataFrame是一种分布式的数据集合，它可以看作是一个表格，其中每一行都有相同的结构，每一列都有一个名称。DataFrame API提供了一系列的操作，包括选择、过滤、聚合等，可以方便地进行数据处理和分析。
总之，Spark SQL提供了一种基于结构化数据的编程接口，支持使用SQL语言进行数据查询和操作，同时也支持使用DataFrame API进行编程。它的编程模型基于RDD模型，但是将RDD的抽象概念扩展到了结构化数据上。