【Spark研究】用Apache Spark进行大数据处理第二部分:Spark SQL

在Apache Spark文章系列的前一篇文章中,我们学习了什么是Apache Spark框架,以及如何用该框架帮助组织处理大数据处理分析的需求。

Spark SQL,作为Apache Spark大数据框架的一部分,主要用于结构化数据处理和对Spark数据执行类SQL的查询。通过Spark SQL,可以针对不同格式的数据执行ETL操作(如JSON,Parquet,数据库)然后完成特定的查询操作。

在这一文章系列的第二篇中,我们将讨论Spark SQL库,如何使用Spark SQL库对存储在批处理文件、JSON数据集或Hive表中的数据执行SQL查询。

Spark大数据处理框架目前最新的版本是上个月发布的Spark 1.3。这一版本之前,Spark SQL模块一直处于“Alpha”状态,现在该团队已经从Spark SQL库上将这一标签移除。这一版本中包含了许多新的功能特性,其中一部分如下:

  • 数据框架(DataFrame):Spark新版本中提供了可以作为分布式SQL查询引擎的程序化抽象DataFrame。
  • 数据源(Data Sources):随着数据源API的增加,Spark SQL可以便捷地处理以多种不同格式存储的结构化数据,如Parquet,JSON以及Apache Avro库。
  • JDBC服务器(JDBC Server):内置的JDBC服务器可以便捷地连接到存储在关系型数据库表中的结构化数据并利用传统的商业智能(BI)工具进行大数据分析。

Spark SQL组件

使用Spark SQL时,最主要的两个组件就是DataFrame和SQLContext。

首先,我们来了解一下DataFrame。

DataFrame

DataFrame是一个分布式的,按照命名列的形式组织的数据集合。DataFrame基于R语言中的data frame概念,与关系型数据库中的数据库表类似。

之前版本的Spark SQL API中的SchemaRDD已经更名为DataFrame。

通过调用将DataFrame的内容作为行RDD(RDD of Rows)返回的rdd方法,可以将DataFrame转换成RDD。

可以通过如下数据源创建DataFrame:

  • 已有的RDD
  • 结构化数据文件
  • JSON数据集
  • Hive表
  • 外部数据库

Spark SQL和DataFrame API已经在下述几种程序设计语言中实现:

  • Scala(https://spark.apache.org/docs/1.3.0/api/scala/index.html#org.apache.spark.sql.package)
  • Java(https://spark.apache.org/docs/1.3.0/api/java/index.html?org/apache/spark/sql/api/java/package-summary.html)
  • Python(https://spark.apache.org/docs/1.3.0/api/python/pyspark.sql.html)

本文中所涉及的Spark SQL代码示例均使用Spark Scala Shell程序。

SQLContext

Spark SQL提供SQLContext封装Spark中的所有关系型功能。可以用之前的示例中的现有SparkContext创建SQLContext。下述代码片段展示了如何创建一个SQLContext对象。

val sqlContext = new org.apache.spark.sql.SQLContext(sc)

此外,Spark SQL中的HiveContext可以提供SQLContext所提供功能的超集。可以在用HiveQL解析器编写查询语句以及从Hive表中读取数据时使用。

在Spark程序中使用HiveContext无需既有的Hive环境。

JDBC数据源

Spark SQL库的其他功能还包括数据源,如JDBC数据源。

JDBC数据源可用于通过JDBC API读取关系型数据库中的数据。相比于使用JdbcRDD,应该将JDBC数据源的方式作为首选,因为JDBC数据源能够将结果作为DataFrame对象返回,直接用Spark SQL处理或与其他数据源连接。

Spark SQL示例应用

在上一篇文章中,我们学习了如何在本地环境中安装Spark框架,如何启动Spark框架并用Spark Scala Shell与其交互。如需安装最新版本的Spark,可以从Spark网站下载该软件。

对于本文中的代码示例,我们将使用相同的Spark Shell执行Spark SQL程序。这些代码示例适用于Windows环境。

为了确保Spark Shell程序有足够的内存,可以在运行spark-shell命令时,加入driver-memory命令行参数,如下所示:

spark-shell.cmd --driver-memory 1G

Spark SQL应用

Spark Shell启动后,就可以用Spark SQL API执行数据分析查询。

在第一个示例中,我们将从文本文件中加载用户数据并从数据集中创建一个DataFrame对象。然后运行DataFrame函数,执行特定的数据选择查询。

文本文件customers.txt中的内容如下:

100, John Smith, Austin, TX, 78727
200, Joe Johnson, Dallas, TX, 75201
300, Bob Jones, Houston, TX, 77028
400, Andy Davis, San Antonio, TX, 78227
500, James Williams, Austin, TX, 78727

下述代码片段展示了可以在Spark Shell终端执行的Spark SQL命令。

// 首先用已有的Spark Context对象创建SQLContext对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 导入语句,可以隐式地将RDD转化成DataFrame
import sqlContext.implicits._

// 创建一个表示客户的自定义类
case class Customer(customer_id: Int, name: String, city: String, state: String, zip_code: String)

// 用数据集文本文件创建一个Customer对象的DataFrame
val dfCustomers = sc.textFile("data/customers.txt").map(_.split(",")).map(p => Customer(p(0).trim.toInt, p(1), p(2), p(3), p(4))).toDF()

// 将DataFrame注册为一个表
dfCustomers.registerTempTable("customers")

// 显示DataFrame的内容
dfCustomers.show()

// 打印DF模式
dfCustomers.printSchema()

// 选择客户名称列
dfCustomers.select("name").show()

// 选择客户名称和城市列
dfCustomers.select("name", "city").show()

// 根据id选择客户
dfCustomers.filter(dfCustomers("customer_id").equalTo(500)).show()

// 根据邮政编码统计客户数量
dfCustomers.groupBy("zip_code").count().show()

在上一示例中,模式是通过反射而得来的。我们也可以通过编程的方式指定数据集的模式。这种方法在由于数据的结构以字符串的形式编码而无法提前定义定制类的情况下非常实用。

如下代码示例展示了如何使用新的数据类型类StructType,StringType和StructField指定模式。

//
// 用编程的方式指定模式
//

// 用已有的Spark Context对象创建SQLContext对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 创建RDD对象
val rddCustomers = sc.textFile("data/customers.txt")

// 用字符串编码模式
val schemaString = "customer_id name city state zip_code"

// 导入Spark SQL数据类型和Row
import org.apache.spark.sql._

import org.apache.spark.sql.types._;

// 用模式字符串生成模式对象
val schema = StructType(schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))

// 将RDD(rddCustomers)记录转化成Row。
val rowRDD = rddCustomers.map(_.split(",")).map(p => Row(p(0).trim,p(1),p(2),p(3),p(4)))

// 将模式应用于RDD对象。
val dfCustomers = sqlContext.createDataFrame(rowRDD, schema)

// 将DataFrame注册为表
dfCustomers.registerTempTable("customers")

// 用sqlContext对象提供的sql方法执行SQL语句。
val custNames = sqlContext.sql("SELECT name FROM customers")

// SQL查询的返回结果为DataFrame对象,支持所有通用的RDD操作。
// 可以按照顺序访问结果行的各个列。
custNames.map(t => "Name: " + t(0)).collect().foreach(println)

// 用sqlContext对象提供的sql方法执行SQL语句。
val customersByCity = sqlContext.sql("SELECT name,zip_code FROM customers ORDER BY zip_code")

// SQL查询的返回结果为DataFrame对象,支持所有通用的RDD操作。
// 可以按照顺序访问结果行的各个列。
customersByCity.map(t => t(0) + "," + t(1)).collect().foreach(println)

除了文本文件之外,也可以从其他数据源中加载数据,如JSON数据文件,Hive表,甚至可以通过JDBC数据源加载关系型数据库表中的数据。

如上所示,Spark SQL提供了十分友好的SQL接口,可以与来自多种不同数据源的数据进行交互,而且所采用的语法也是团队熟知的SQL查询语法。这对于非技术类的项目成员,如数据分析师以及数据库管理员来说,非常实用。

总结

本文中,我们了解到Apache Spark SQL如何用熟知的SQL查询语法提供与Spark数据交互的SQL接口。Spark SQL是一个功能强大的库,组织中的非技术团队成员,如业务分析师和数据分析师,都可以用Spark SQL执行数据分析。

下一篇文章中,我们将讨论可用于处理实时数据或流数据的Spark Streaming库。Spark Streaming库是任何一个组织的整体数据处理和管理生命周期中另外一个重要的组成部分,因为流数据处理可为我们提供对系统的实时观察。这对于欺诈检测、在线交易系统、事件处理解决方案等用例来说至关重要。

参考文献

  • Spark主站
  • Spark SQL网站
  • Spark SQL程序设计指南
  • 用Apache Spark进行大数据处理——第一部分:入门介绍

来源:http://www.infoq.com/cn/articles/apache-spark-sql

原文发布于微信公众号 - 数据科学与人工智能(DS_AI_shujuren)

原文发表时间:2015-12-15

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Albert陈凯

3.6 Shuffle机制

3.6 Shuffle机制 在MapReduce框架中,Shuffle是连接Map和Reduce之间的桥梁,Map的输出要用到Reduce中必须经过Shuff...

2914
来自专栏xingoo, 一个梦想做发明家的程序员

Structured Streaming教程(1) —— 基本概念与使用

在有过1.6的streaming和2.x的streaming开发体验之后,再来使用Structured Streaming会有一种完全不同的体验,尤其是在代码设...

1421
来自专栏Hadoop实操

Spark2Streaming读Kerberos环境的Kafka并写数据到Kudu

1K3
来自专栏Hadoop实操

Spark2Streaming读非Kerberos环境的Kafka并写数据到Kudu

在前面的文章Fayson介绍了在Kerberos环境下《Spark2Streaming读Kerberos环境的Kafka并写数据到Kudu》,本篇文章Fayso...

3521
来自专栏Hadoop实操

SparkStreaming读Kafka数据写Kudu

1.9K2
来自专栏about云

spark零基础学习线路指导【包括spark2】

问题导读 1.你认为spark该如何入门? 2.你认为spark入门编程需要哪些步骤? 3.本文介绍了spark哪些编程知识?

1053
来自专栏Java技术栈

Tomcat集群session复制与Oracle的坑。。

问题描述 公司某个系统使用了tomcat自带的集群session复制功能,然后后报了一个oracle驱动包里面的连接不能被序列化的异常。 01-Nov-2017...

3779
来自专栏Hadoop实操

Spark2Streaming读Kerberos环境的Kafka并写数据到Hive

在前面的文章Fayson介绍了一些关于Spark2Streaming的示例如《Spark2Streaming读Kerberos环境的Kafka并写数据到HBas...

1K3
来自专栏祝威廉

Spark 2.0 Structured Streaming 分析

Spark 2.0 将流式计算也统一到DataFrame里去了,提出了Structured Streaming的概念,将数据源映射为一张无线长度的表,同时将流式...

1383
来自专栏知识分享

串口通信DMA中断

这是以前学32的时候写的,那时候学了32之后感觉32真是太强大了,比51强的没影。关于dma网上有许多的资料,亲们搜搜,这里只贴代码了,其实我也想详详细细地叙述...

3247

扫码关注云+社区

领取腾讯云代金券