spark2 sql读取数据源编程学习样例2:函数实现详解

本文涉及的产品
云原生数据库 PolarDB MySQL 版,Serverless 5000PCU 100GB
云原生数据库 PolarDB PostgreSQL 版,企业版 4核16GB
推荐场景:
HTAP混合负载
云原生数据库 PolarDB MySQL 版,通用型 2核4GB 50GB
简介: spark2 sql读取数据源编程学习样例2:函数实现详解

这里接着上篇,继续阅读代码,下面我们看看runBasicParquetExample函数的功能实现

runBasicParquetExample函数



private def runBasicParquetExample(spark: SparkSession): Unit = {
   // $example on:basic_parquet_example$
   // Encoders for most common types are automatically provided by importing spark.implicits._
   import spark.implicits._
   val peopleDF = spark.read.json("examples/src/main/resources/people.json")
   // DataFrames can be saved as Parquet files, maintaining the schema information
   peopleDF.write.parquet("people.parquet")
   // Read in the parquet file created above
   // Parquet files are self-describing so the schema is preserved
   // The result of loading a Parquet file is also a DataFrame
   val parquetFileDF = spark.read.parquet("people.parquet")
   // Parquet files can also be used to create a temporary view and then used in SQL statements
   parquetFileDF.createOrReplaceTempView("parquetFile")
   val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")
   namesDF.map(attributes => "Name: " + attributes(0)).show()
   // +------------+
   // |       value|
   // +------------+
   // |Name: Justin|
   // +------------+
   // $example off:basic_parquet_example$
 }

这里面有一个包的导入

import spark.implicits._

Scala中与其它语言的区别是在对象,函数中可以导入包。这个包的作用是转换RDD为DataFrame。

val peopleDF = spark.read.json("examples/src/main/resources/people.json")

上面自然是读取json文件。

peopleDF.write.parquet("people.parquet")


这里同样是保存文件,不过people.parquet是文件夹。文件夹里面是数据,其中有*00000*为数据文件。

val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")

这里调用sql语句。

namesDF.map(attributes => "Name: " + attributes(0)).show()

这里通过map映射,增加Name:

// +------------+
    // |       value|
    // +------------+
    // |Name: Justin|
    // +------------+
    // $example off:basic_parquet_example$




runParquetSchemaMergingExample函数


private def runParquetSchemaMergingExample(spark: SparkSession): Unit = {
    // $example on:schema_merging$
    // This is used to implicitly convert an RDD to a DataFrame.
    import spark.implicits._
    // Create a simple DataFrame, store into a partition directory
    val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")
    squaresDF.write.parquet("data/test_table/key=1")
    // Create another DataFrame in a new partition directory,
    // adding a new column and dropping an existing column
    val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")
    cubesDF.write.parquet("data/test_table/key=2")
    // Read the partitioned table
    val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
    mergedDF.printSchema()
    // The final schema consists of all 3 columns in the Parquet files together
    // with the partitioning column appeared in the partition directory paths
    // root
    //  |-- value: int (nullable = true)
    //  |-- square: int (nullable = true)
    //  |-- cube: int (nullable = true)
    //  |-- key: int (nullable = true)
    // $example off:schema_merging$
  }


val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")
squaresDF.write.parquet("data/test_table/key=1")

上面是创建一个RDD,然后通过toDF转换为DataFrame。然后保存到分区目录下。

val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")
  cubesDF.write.parquet("data/test_table/key=2")

创建另外一个DataFrame,并且添加一个新列,删除现有列

val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
   mergedDF.printSchema()

上面自然是读取数据保存为DataFrame,option("mergeSchema", "true"), 默认值由spark.sql.parquet.mergeSchema指定。设置所有的分区文件是否合并Schema。设置后将覆盖spark.sql.parquet.mergeSchema指定值。


runJsonDatasetExample函数



private def runJsonDatasetExample(spark: SparkSession): Unit = {
    // $example on:json_dataset$
    // Primitive types (Int, String, etc) and Product types (case classes) encoders are
    // supported by importing this when creating a Dataset.
    import spark.implicits._
    // A JSON dataset is pointed to by path.
    // The path can be either a single text file or a directory storing text files
    val path = "examples/src/main/resources/people.json"
    val peopleDF = spark.read.json(path)
    // The inferred schema can be visualized using the printSchema() method
    peopleDF.printSchema()
    // root
    //  |-- age: long (nullable = true)
    //  |-- name: string (nullable = true)
    // Creates a temporary view using the DataFrame
    peopleDF.createOrReplaceTempView("people")
    // SQL statements can be run by using the sql methods provided by spark
    val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
    teenagerNamesDF.show()
    // +------+
    // |  name|
    // +------+
    // |Justin|
    // +------+
    // Alternatively, a DataFrame can be created for a JSON dataset represented by
    // a Dataset[String] storing one JSON object per string
    val otherPeopleDataset = spark.createDataset(
      """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)
    val otherPeople = spark.read.json(otherPeopleDataset)
    otherPeople.show()
    // +---------------+----+
    // |        address|name|
    // +---------------+----+
    // |[Columbus,Ohio]| Yin|
    // +---------------+----+
    // $example off:json_dataset$
  }

上面有些代码重复了,就不在解释了。

val path = "examples/src/main/resources/people.json"    val peopleDF = spark.read.json(path)
    // The inferred schema can be visualized using the printSchema() method
    peopleDF.printSchema()
    // root
    //  |-- age: long (nullable = true)
    //  |-- name: string (nullable = true)

上面是读取json,并显示schema。

// Creates a temporary view using the DataFrame
    peopleDF.createOrReplaceTempView("people")
    // SQL statements can be run by using the sql methods provided by spark
    val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
    teenagerNamesDF.show()

peopleDF.createOrReplaceTempView("people")是DataFrame注册为people表

teenagerNamesDF.show()

自然是显示数据。

如下

// +------+
  // |  name|
  // +------+
  // |Justin|
  // +------+


val otherPeopleDataset = spark.createDataset(
"""{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil)

这里创建一个json格式的dataset

val otherPeople = spark.read.json(otherPeopleDataset)

这行代码,是读取上面创建的dataset,然后创建DataFrame。从上面我们看出这也是dataset和DataFrame转换的一种方式。


runJdbcDatasetExample函数


private def runJdbcDatasetExample(spark: SparkSession): Unit = {
    // $example on:jdbc_dataset$
    // Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
    // Loading data from a JDBC source
    val jdbcDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:postgresql:dbserver")
      .option("dbtable", "schema.tablename")
      .option("user", "username")
      .option("password", "password")
      .load()
    val connectionProperties = new Properties()
    connectionProperties.put("user", "username")
    connectionProperties.put("password", "password")
    val jdbcDF2 = spark.read
      .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
    // Specifying the custom data types of the read schema
    connectionProperties.put("customSchema", "id DECIMAL(38, 0), name STRING")
    val jdbcDF3 = spark.read
      .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
    // Saving data to a JDBC source
    jdbcDF.write
      .format("jdbc")
      .option("url", "jdbc:postgresql:dbserver")
      .option("dbtable", "schema.tablename")
      .option("user", "username")
      .option("password", "password")
      .save()
    jdbcDF2.write
      .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
    // Specifying create table column data types on write
    jdbcDF.write
      .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")
      .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
    // $example off:jdbc_dataset$
  }
}

这个是运行Jdbc Dataset的例子。那么如何从jdbc读取数据,是通过下面各个option

.option("url", "jdbc:postgresql:dbserver")
     .option("dbtable", "schema.tablename")
     .option("user", "username")
     .option("password", "password")

第一行server也就是服务器地址

第二行是表名

第三行是用户名

第四行为密码,相信大家也能看明白。

val connectionProperties = new Properties()

Properties这个是用来做什么的那?

我们来看官网

c4510f216b1cc6479d9f8487268a97c4.jpg

它是

JDBC database 连接的一个参数,是一个字符串tag/value的列表。于是有了下面内容

connectionProperties.put("user", "username")
   connectionProperties.put("password", "password")

我们看到上面放入了用户名和密码

val jdbcDF2 = spark.read
      .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

这里设置了连接url,表名,还有connectionProperties

connectionProperties.put("customSchema", "id DECIMAL(38, 0), name STRING")

上面是指定读取Schema的自定义数据类型。

jdbcDF.write
   .format("jdbc")
   .option("url", "jdbc:postgresql:dbserver")
   .option("dbtable", "schema.tablename")
   .option("user", "username")
   .option("password", "password")
   .save()
jdbcDF2.write
   .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
 // Specifying create table column data types on write
 jdbcDF.write
   .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")
   .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

上面分别都是将数据通过jdbc保存到数据库


相关实践学习
使用PolarDB和ECS搭建门户网站
本场景主要介绍基于PolarDB和ECS实现搭建门户网站。
阿里云数据库产品家族及特性
阿里云智能数据库产品团队一直致力于不断健全产品体系,提升产品性能,打磨产品功能,从而帮助客户实现更加极致的弹性能力、具备更强的扩展能力、并利用云设施进一步降低企业成本。以云原生+分布式为核心技术抓手,打造以自研的在线事务型(OLTP)数据库Polar DB和在线分析型(OLAP)数据库Analytic DB为代表的新一代企业级云原生数据库产品体系, 结合NoSQL数据库、数据库生态工具、云原生智能化数据库管控平台,为阿里巴巴经济体以及各个行业的企业客户和开发者提供从公共云到混合云再到私有云的完整解决方案,提供基于云基础设施进行数据从处理、到存储、再到计算与分析的一体化解决方案。本节课带你了解阿里云数据库产品家族及特性。
目录
相关文章
|
19天前
|
SQL 数据库
SQL FORMAT() 函数
SQL FORMAT() 函数
19 2
|
29天前
|
SQL XML 关系型数据库
【less-5】基于SQLI的SQL盲注常用函数
【less-5】基于SQLI的SQL盲注常用函数
28 2
|
22天前
|
SQL 数据库
SQL SUM() 函数
SQL SUM() 函数
17 0
|
21天前
|
SQL 数据库
SQL LCASE() 函数
SQL LCASE() 函数
19 7
|
21天前
|
SQL 数据库
SQL UCASE() 函数
SQL UCASE() 函数
16 4
|
20天前
|
SQL
SQL ROUND() 函数
SQL ROUND() 函数
21 1
|
24天前
|
SQL Oracle 关系型数据库
SQL LAST() 函数
SQL LAST() 函数
18 5
|
25天前
|
SQL Oracle 关系型数据库
SQL FIRST() 函数
SQL FIRST() 函数
15 3
|
25天前
|
SQL Oracle 关系型数据库
SQL COUNT() 函数
SQL COUNT() 函数
15 3
|
25天前
|
SQL 数据库
SQL AVG() 函数
SQL AVG() 函数
14 2