写点什么

用 Apache Spark 进行大数据处理——第二部分:Spark SQL

  • 2015-06-12
  • 本文字数:3896 字

    阅读完需:约 13 分钟

在 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 已经在下述几种程序设计语言中实现:

本文中所涉及的 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 库是任何一个组织的整体数据处理和管理生命周期中另外一个重要的组成部分,因为流数据处理可为我们提供对系统的实时观察。这对于欺诈检测、在线交易系统、事件处理解决方案等用例来说至关重要。

参考文献

关于作者

Srini Penchikala目前是一家金融服务机构的软件架构师,这个机构位于德克萨斯州的奥斯汀。他在软件系统架构、设计和开发方面有超过 20 年的经验。Srini 目前正在撰写一本关于 NoSQL 数据库模式的书。他还是曼宁出版社出版的《Spring Roo in Action》一书的合著者( http://www.manning.com/SpringRooinAction)。他还曾经出席各种会议,如 JavaOne,SEI Architecture Technology Conference(SATURN),IT Architect Conference(ITARC),No Fluff Just Stuff,NoSQL Now 和 Project World Conference 等。Srini 还在 InfoQ,The ServerSide,OReilly Network(ONJava),DevX Java,java.net 以及 JavaWorld 等网站上发表过很多关于软件系统架构、安全和风险管理以及 NoSQL 数据库等方面的文章。他还是 InfoQ NoSQL 数据库社区的责任编辑( http://www.infoq.com/author/Srini-Penchikala)。

查看英文原文: Big Data Processing with Apache Spark - Part 2: Spark SQL

公众号推荐:

2024 年 1 月,InfoQ 研究中心重磅发布《大语言模型综合能力测评报告 2024》,揭示了 10 个大模型在语义理解、文学创作、知识问答等领域的卓越表现。ChatGPT-4、文心一言等领先模型在编程、逻辑推理等方面展现出惊人的进步,预示着大模型将在 2024 年迎来更广泛的应用和创新。关注公众号「AI 前线」,回复「大模型报告」免费获取电子版研究报告。

AI 前线公众号
2015-06-12 03:4253634
用户头像

发布了 75 篇内容, 共 61.9 次阅读, 收获喜欢 6 次。

关注

评论

发布
暂无评论
发现更多内容

apm应用性能管理,非科班程序员求职经历分享,看这一篇就够了

欢喜学安卓

android 程序员 面试 移动开发

【实战问题】-- 布隆过滤器的三种实践:手写,Redission以及Guava(2)

秦怀杂货店

Java 布隆过滤器

数据产品经理的圣经,送你一份真贵的礼物 ~~ 年度数据产品经理们的总结

松子(李博源)

大数据 数据产品经理 数据产品

hive的主流文件存储格式对比实验

大数据技术指南

大数据 hive 5月日更

Android屏幕适配,音视频学习指南来咯,文末领取面试资料

欢喜学安卓

android 程序员 面试 移动开发

引荐好友成为推广者还能拿额外奖励?!华为云引荐奖励计划来啦!

华为云开发者联盟

文章 返现奖励 推广计划 返利 团长

HTML回流与重绘

空城机

JavaScript html 大前端 5月日更

微服务化转型,拆就行了?这样做很危险...

BoCloud博云

微服务

博睿数据携手阿里云,打造以用户为中心的智能运维生态

博睿数据

阿里云 智能运维 博睿数据

XDPool比特兄弟矿场系统开发|XDPool比特兄弟矿场APP软件开发

NetWebCore实现文件上传功能

happlyfox

学习 .net core 5月日更

看亚马逊云科技如何助力英国教育科技公司 Firefly 快速落地中国|精选案例

亚马逊云科技 (Amazon Web Services)

精选Hadoop高频面试题17道,附答案详细解析

五分钟学大数据

大数据 hadoop 5月日更

这是我金三银四收到的第6个Offer:美团+阿里Java研发岗

Java 程序员 架构 面试

差点败北,蚂蚁金服的六轮面试我是怎么撑过来的?(Java岗)

Java架构师迁哥

看Amazon Graviton 如何实现优化性能、降低工作负载成本!

亚马逊云科技 (Amazon Web Services)

Web渗透测试:信息收集篇

Thrash

antd table input 失焦的问题

blueju

打破思维定式(八)

Changing Lin

如何从一段视频中一次性修整多个片段

奈奈的杂社

视频剪辑 视频后期 视频处理

v03.06 鸿蒙内核源码分析(时钟任务) | 触发调度谁的贡献最大 | 百篇博客分析HarmonyOS源码

鸿蒙研究站

鸿蒙内核源码分析 百篇博客分析鸿蒙

2021年4月云主机性能评测报告新鲜出炉,盛大云华东蝉联冠军!

博睿数据

一文掌握 Java8 Stream 中 Collectors 的 24 个操作

看山

Java

NAT穿透原理详解

IT酷盖

音视频 p2p NAT

兄弟矿场系统开发|兄弟矿场软件APP开发

交通流量预测,EasyDL带你零代码实战

百度大脑

零代码 EasyDL

Iceberg0.11与Spark3.0结合

InfoQ_Springup

iceberg

零基础学习 NLP-DAY4 - 恶补Python - 基础1

Qien Z.

Python nlp 5月日更

HashMap

ltc

hashmap

这个好用的分布式应用配置中心,我们把它开源了

百度Geek说

分布式 大前端 服务器

iOS 面试策略之语言工具-Xcode使用

iOSer

ios xcode 语言 & 开发

用Apache Spark进行大数据处理——第二部分:Spark SQL_DevOps & 平台工程_Srini Penchikala_InfoQ精选文章