Spark SQL实战(04)-API编程之DataFrame

这篇具有很好参考价值的文章主要介绍了Spark SQL实战(04)-API编程之DataFrame。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1 SparkSession

Spark Core: SparkContext

Spark SQL: 难道就没有SparkContext?

2.x之后统一的

package com.javaedge.bigdata.chapter04

import org.apache.spark.sql.{DataFrame, SparkSession}

object SparkSessionApp {

  def main(args: Array[String]): Unit = {

    // DF/DS编程的入口点
    val spark: SparkSession = SparkSession.builder()
      .master("local").getOrCreate()

    // 读取文件的API
    val df: DataFrame = spark.read.text("/Users/javaedge/Downloads/sparksql-train/data/input.txt")

    // TODO 业务逻辑处理,通过DF/DS提供的API完成业务
    df.printSchema()
    df.show() // 展示出来  只有一个字段,string类型的value

    spark.stop()
  }
}

1.x的Spark SQL编程入口点

  • SQLContext
  • HiveContext

Spark SQL中,SQLContext、HiveContext都是用来创建DataFrame和Dataset主要入口点,二者区别如下:

  1. 数据源支持:SQLContext支持的数据源包括JSON、Parquet、JDBC等等,而HiveContext除了支持SQLContext的数据源外,还支持Hive的数据源。因此,如果需要访问Hive中的数据,需要使用HiveContext。
  2. 元数据管理:SQLContext不支持元数据管理,因此无法在内存中创建表和视图,只能直接读取数据源中的数据。而HiveContext可以在内存中创建表和视图,并将其存储在Hive Metastore中。
  3. SQL语言支持:SQLContext和HiveContext都支持Spark SQL中的基本语法,例如SELECT、FROM、WHERE等等。但HiveContext还支持Hive中的所有SQL语法,例如INSERT、CREATE TABLE AS等等。
  4. 数据格式支持:HiveContext支持更多的数据格式,包括ORC、Avro、SequenceFile等等。而SQLContext只支持JSON、Parquet、JDBC等几种常用的数据格式。

如若访问Hive中数据或在内存中创建表和视图,推荐HiveContext;若只需访问常见数据源,使用SQLContext。

Spark 2.x后,HiveContext已被SparkSession替代,因此推荐SparkSession创建DataFrame、Dataset。

package com.javaedge.bigdata.chapter04

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

/**
 * 了解即可,已过时
 */
object SQLContextApp {

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("SQLContextApp").setMaster("local")
    // 此处一定要把SparkConf传进来
    val sc: SparkContext = new SparkContext(sparkConf)
    val sqlContext: SQLContext = new SQLContext(sc)

    val df: DataFrame = sqlContext.read.text("/Users/javaedge/Downloads/sparksql-train/data/input.txt")
    df.show()

    sc.stop()
  }
}


output:
+-------------+
|        value|
+-------------+
|     pk,pk,pk|
|jepson,jepson|
|     xingxing|
+-------------+

2 DataFrame

最早在R语言数据分析包中提出,表示一种类似表格的数据结构,其中行和列都可以有命名。

Spark的DataFrame是基于RDD(弹性分布式数据集)的一种高级抽象,类似关系型数据库的表格。Spark DataFrame可看作带有模式(Schema)的RDD,而Schema则是由结构化数据类型(如字符串、整型、浮点型等)和字段名组成。

2.1 命名变迁

Spark 1.0的Spark SQL的数据结构称为SchemaRDD,具有结构化模式(schema)的分布式数据集合。

Spark 1.3版本开始,SchemaRDD重命名为DataFrame,以更好反映其API和功能实质。因此,DataFrame曾被称为SchemaRDD,但现已不再使用这名称。

2.2 Spark SQL的DataFrame优点

  • 可通过SQL语句、API等多种方式进行查询和操作,还支持内置函数、用户自定义函数等功能
  • 支持优化器和执行引擎,可自动对查询计划进行优化,提高查询效率

因此,DataFrame已成Spark SQL核心组件,广泛应用于数据分析、数据挖掘。

3 数据分析选型:PySpark V.S R 语言

  1. 数据规模:如果需要处理大型数据集,则使用PySpark更为合适,因为它可以在分布式计算集群上运行,并且能够处理较大规模的数据。而R语言则可能会受限于单机内存和计算能力。
  2. 熟练程度:如果你或你的团队已经很熟悉Python,那么使用PySpark也许更好一些,因为你们不需要再去学习新的编程语言。相反,如果已经对R语言很熟悉,那么继续使用R语言也许更为方便。
  3. 生态系统:Spark生态系统提供了许多额外的库和工具,例如Spark Streaming和GraphX等,这些库和工具可以与PySpark无缝集成。而R语言的生态系统也有一些类似的库和工具,但相对来说可选择性就更少一些。

总之,选择使用哪种工具进行数据分析应该基于具体情况进行考虑。如果需要处理大规模数据集,并需要与Spark生态系统集成,那么PySpark可能更适合;如果更加熟悉R语言,或者数据量较小,那么使用R语言也可以做到高效的数据分析。

4 深入理解

Dataset是一个分布式数据集,提供RDD强类型和使用强大的lambda函数的能力,并结合了Spark SQL优化的执行引擎。Dataset可以从JVM对象构建而成,并通过函数式转换(如map、flatMap、filter等)进行操作。Scala和Java都支持Dataset API,但Python没有对Dataset API提供支持。由于Python是一种动态语言,许多Dataset API的优点已经自然地可用,例如可以通过名称访问行的字段。R语言也有类似的特点。

DataFrame,具有命名列的Dataset,类似:

  • 关系数据库中的表
  • Python中的数据框

但内部有更多优化功能。DataFrame可从各种数据源构建,如:

  • 结构化数据文件
  • Hive表
  • 外部数据库
  • 现有RDD

DataFrame API 在 Scala、Java、Python 和 R 都可用。在Scala和Java中,DataFrame由一组Rows组成的Dataset表示:

  • Scala API中,DataFrame只是Dataset[Row]的类型别名
  • Java API中,用户需要使用Dataset表示DataFrame

通常将Scala/Java中的Dataset of Rows称为DataFrame。

5 实战

People.json

{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}
package com.javaedge.bigdata.chapter04

import org.apache.spark.sql.{DataFrame, SparkSession}

object DataFrameAPIApp {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .master("local").appName("DataFrameAPIApp")
      .getOrCreate()
    import spark.implicits._


    val people: DataFrame = spark.read.json(
      "/Users/javaedge/Downloads/sparksql-train/data/people.json")

    // 查看DF的内部结构:列名、列的数据类型、是否可以为空
    people.printSchema()

    // 展示出DF内部的数据
    people.show()
  }
}


output:
root
  |-- age: long (nullable = true)
  |-- name: string (nullable = true)

+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+
// DF里面有两列,只要name列 ==> select name from people
// 两个 API 一样的,只是参数不同,使用稍有不同
people.select("name").show()
people.select($"name").show()

output:

+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+
// select * from people where age > 21
people.filter($"age" > 21).show()
people.filter("age > 21").show()


output:
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+
// select age, count(1) from people group by age
people.groupBy("age").count().show()

output:
+----+-----+
| age|count|
+----+-----+
|  19|    1|
|null|    1|
|  30|    1|
+----+-----+

createOrReplaceTempView

若现在,我就想完全使用 SQL 查询了,怎么实现 DF 到表的转换呢?

Spark SQL用来将一个 DataFrame 注册成一个临时表(Temporary Table)的方法。之后可使用 Spark SQL 语法及已注册的表名对 DataFrame 进行查询和操作。

允许为 DataFrame 指定一个名称,并将其保存为一个临时表。该表只存在于当前 SparkSession 的上下文,不会在元数据存储中注册表,也不会在磁盘创建任何文件。因此,临时表在SparkSession终止后就会被删。

一旦临时表被注册,就可使用 SQL 或 DSL 对其查询。如:

people.createOrReplaceTempView("people")

spark.sql("select name from people where age > 21").show()

大文件处理

val zips: DataFrame = spark.read.json("/Users/javaedge/Downloads/sparksql-train/data/zips.json")
zips.printSchema()

zips.show(5)

loc信息没用展示全,超过一定长度就使用…来展示,默认只显示前20条:show() ==> show(20) ==> show(numRows, truncate = true)

Spark SQL实战(04)-API编程之DataFrame

不想被截断就这样:

Spark SQL实战(04)-API编程之DataFrame

zips.head(3).foreach(println)
zips.first()
zips.take(5)

Output:

Spark SQL实战(04)-API编程之DataFrame

head(n: Int)

Spark的DataFrame API中的一个方法,可以返回一个包含前n行数据的数组。这个方法通常用于快速检查一个DataFrame的前几行数据,以了解数据集的大致结构和内容。

  • 先对DataFrame使用.limit(n)方法,限制返回行数前n行
  • 然后使用queryExecution方法生成一个Spark SQL查询计划
  • 最后使用collectFromPlan方法收集数据并返回一个包含前n行数据的数组

该 API 可能导致数据集的全部数据被加载到内存,因此在处理大型数据集时应该谨慎使用。若只想查看数据集结构和内容,使用:

  • .show()
  • .take()

获取前几行数据,而非.head()

// 过滤出大于40000,字段重新命名
zips.filter(zips.col("pop") > 40000)
  .withColumnRenamed("_id", "new_id")
  .show(5, truncate = false)
output:
+------+----------+-----------------------+-----+-----+
|new_id|city      |loc                    |pop  |state|
+------+----------+-----------------------+-----+-----+
|01040 |HOLYOKE   |[-72.626193, 42.202007]|43704|MA   |
|01085 |MONTGOMERY|[-72.754318, 42.129484]|40117|MA   |
|01201 |PITTSFIELD|[-73.247088, 42.453086]|50655|MA   |
|01420 |FITCHBURG |[-71.803133, 42.579563]|41194|MA   |
|01701 |FRAMINGHAM|[-71.425486, 42.300665]|65046|MA   |
+------+----------+-----------------------+-----+-----+


import org.apache.spark.sql.functions._
// 统计加州pop最多的10个城市名称和ID  desc是一个内置函数
zips.select("_id", "city", "pop", "state")
  .filter(zips.col("state") === "CA")
  .orderBy(desc("pop"))
  .show(5, truncate = false)
output:
+-----+------------+-----+-----+
|_id  |city        |pop  |state|
+-----+------------+-----+-----+
|90201|BELL GARDENS|99568|CA   |
|90011|LOS ANGELES |96074|CA   |
|90650|NORWALK     |94188|CA   |
|91331|ARLETA      |88114|CA   |
|90280|SOUTH GATE  |87026|CA   |
+-----+------------+-----+-----+

可惜啊,我不会写代码,可以使用 MySQL 语法吗?

zips.createOrReplaceTempView("zips")
spark.sql("select _id,city,pop,state" +
  "from zips where state='CA'" +
  "order by pop desc" +
  "limit 10").show()

import spark.implicits._ 作用

在Scala中使用Apache Spark进行数据分析时经常用到的,它的作用是将隐式转换函数导入当前作用域中。这些隐式转换函数包含了许多DataFrame和Dataset的转换方法,例如将RDD转换为DataFrame或将元组转换为Dataset等。

具体来说,这行代码使用了SparkSession对象中的implicits属性,该属性返回了一个类型为org.apache.spark.sql.SQLImplicits的实例。通过调用该实例的方法,可以将各种Scala数据类型(如case class、元组等)与Spark SQL中的数据类型(如Row、DataFrame、Dataset等)之间进行转换,从而方便地进行数据操作和查询。

在使用许多Spark SQL API的时候,往往需要使用这行代码将隐式转换函数导入当前上下文,以获得更加简洁和易于理解的代码编写方式。

如果不导入会咋样

如果不导入spark.implicits._会导致编译错误或者运行时异常。因为在进行DataFrame和Dataset的操作时,需要使用到一些隐式转换函数。如果没有导入spark.implicits._,则这些隐式转换函数无法被自动引入当前上下文,就需要手动地导入这些函数,这样会使编码变得比较麻烦。

例如,在进行RDD和DataFrame之间的转换时,如果不导入spark.implicits._,则需要手动导入org.apache.spark.sql.Row以及org.apache.spark.sql.functions._等包,并通过调用toDF()方法将RDD转换为DataFrame。而有了导入spark.implicits._后,只需要直接调用RDD对象的toDF()方法即可完成转换。

因此,为了简化编码,通常会在Scala中使用Spark SQL时导入spark.implicits._,从而获得更加简洁易读的代码。

案例
people.select($"name").show()

如果不导入 spark.implicits._,则可以手动创建一个 Column 对象来进行筛选操作。例如,可以使用 col 函数来创建一个 Column 对象,然后在 select 方法中使用该列:

import org.apache.spark.sql.functions.col

val selected = people.select(col("name"))
selected.show()

这样就可以实现与 people.select($"name").show() 相同的效果,但需要手动创建 Column 对象。显然,在编写复杂的数据操作时,手动创建 Column 对象可能会变得非常繁琐和困难,因此通常情况下我们会选择使用隐式转换函数,从而更加方便地使用DataFrame的API。文章来源地址https://www.toymoban.com/news/detail-405463.html

到了这里,关于Spark SQL实战(04)-API编程之DataFrame的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 大数据Spark SparkSession的3种创建方式 Scala语言实现

    SparkSession是Apache Spark 2.0版本引入的一个编程接口,用于与Spark进行交互。它是Spark应用程序的入口点,提供了一种方便的方式来创建DataFrame、DataSet和SQLContext等数据结构,并且可以配置各种Spark应用程序的选项。SparkSession还管理了Spark应用程序的运行环境,包括Spark集群的连接,

    2023年04月20日
    浏览(38)
  • Spark编程实验三:Spark SQL编程

    目录 一、目的与要求 二、实验内容 三、实验步骤 1、Spark SQL基本操作 2、编程实现将RDD转换为DataFrame 3、编程实现利用DataFrame读写MySQL的数据 四、结果分析与实验体会 1、通过实验掌握Spark SQL的基本编程方法; 2、熟悉RDD到DataFrame的转化方法; 3、熟悉利用Spark SQL管理来自不同

    2024年02月03日
    浏览(39)
  • 15 | Spark SQL 的 SQL API 操作

    SQL API :Spark SQL 允许使用标准 SQL 语句来查询和分析数据。用户可以通过 SparkSession 执行 SQL 查询,并将结果返回为 DataFrame。这使得熟悉 SQL 的用户能够方便地使用 Spark SQL 进行数据处理。 示例 1: 基本查询 执行基本的 SQL 查询,选择数据中的特定列并过滤数据。

    2024年02月09日
    浏览(31)
  • Spark大数据分析与实战笔记(第二章 Spark基础-04)

    “春风十里,不如你。” 这句来自现代作家安妮宝贝的经典句子,它表达了对他人的赞美与崇拜。每个人都有着不同的闪光点和特长,在这个世界上,不必去羡慕别人的光芒,自己所拥有的价值是独一无二的。每个人都有无限的潜力和能力,只要勇敢展现自己,就能在人生舞

    2024年02月03日
    浏览(73)
  • 实训笔记——Spark SQL编程

    Spark SQL属于Spark计算框架的一部分,是专门负责结构化数据的处理计算框架,Spark SQL提供了两种数据抽象:DataFrame、Dataset,都是基于RDD之上的一种高级数据抽象,在RDD基础之上增加了一个schema表结构。 1.1 创建Spark SQL的编程项目,scala语言支持的 1.2 引入编程依赖: spark_core_2.

    2024年02月07日
    浏览(46)
  • 实验三 Spark SQL基础编程

    1. 掌握 Spark SQL 的基本编程方法; 2. 熟悉 RDD 到 DataFrame 的转化方法; 3. 熟悉利用 Spark SQL 管理来自不同数据源的数据。 1.Spark SQL 基本操作 将下列 JSON 格式数据复制到 Linux 系统中,并保存命名为 employee.json。 { \\\"id\\\":1, \\\"name\\\":\\\"Ella\\\", \\\"age\\\":36 } { \\\"id\\\":2, \\\"name\\\":\\\"Bob\\\", \\\"age\\\":29 } { \\\"id\\\":3, \\\"na

    2024年02月06日
    浏览(39)
  • Spark SQL调优实战

    1、 新添参数说明 // D river 和Executor内存和CPU资源相关配置 -- 是否开启 executor 动态分配 , 开启时 spark.executor.instances 不生效 spark.dynamicAllocation.enabled= false --配置Driver内存 spark.dirver.memory=5g --driver最大结果大小,设置为0代表不限制,driver在拉取结果时,如果结果超过阈值会报异

    2024年02月21日
    浏览(31)
  • Spark2.X通过SparkSession读取JDBC数据时遇到Janino库的版本不兼容的错误 org/codehaus/janino/InternalCompilerException

    通过Springboot + Mybatis-plus框架,通过实体类+ Mapper和配置SQL语句的方式读取数据后传递给Spark进行数据分析,一直没有出现问题。 后来需求的不断变化,特别是进行数据分析的时候。数据库的要分析的表字段不断变化,测试表与真实线上表结构也不一样,主要是在字段上。表名

    2024年01月18日
    浏览(65)
  • Spark SQL实战(08)-整合Hive

    Apache Spark 是一个快速、可扩展的分布式计算引擎,而 Hive 则是一个数据仓库工具,它提供了数据存储和查询功能。在 Spark 中使用 Hive 可以提高数据处理和查询的效率。 场景 历史原因积累下来的,很多数据原先是采用Hive来进行处理的,现想改用Spark操作数据,须要求Spark能够

    2023年04月15日
    浏览(92)
  • Spark SQL实战(07)-Data Sources

    Spark SQL通过DataFrame接口支持对多种数据源进行操作。 DataFrame可使用关系型变换进行操作,也可用于创建临时视图。将DataFrame注册为临时视图可以让你对其数据运行SQL查询。 本节介绍使用Spark数据源加载和保存数据的一般方法,并进一步介绍可用于内置数据源的特定选项。 数

    2023年04月08日
    浏览(42)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包