spark sql 语句 字符串转date spark sql字符串拼接

文章目录

• Spark SQL 编程基础

• 1、SQLContext

• 创建 SQLContext

• 2、HiveContext
• 3、DataFrame 编程实例
• 4、创建DataFrame
• 5、SQL 语法
• 6、DSL 语法
• 7、DataFrame 转换为 RDD

Spark SQL 编程基础

在使用Spark-SQL的时候，需要将Hive配置文件目录下的hive-site.xml复制到Spark的conf目录下。

1、SQLContext

Spark SQL 提供了所有功能编程使用的入口点：SQLContext类或者该类的子类。

创建 SQLContext

val sc:SparkContext	// 已经存在的 SparkContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

2、HiveContext

在 spark2.x中，SparkSQL将HiveContext和SQLContext合并，直接使用SparkSession即可。

可以通过spark.sql.dialect选项更改SQL解析器，这个参数在SQLContext的setConf方法中设置，也可以通过SQL的key=value语法设置

3、DataFrame 编程实例

1. 开发环境准备
   在项目的 pom.xml 中添加 Maven 的依赖

<!--Spark SQL 基础依赖-->
<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.10</artifactId>
        <version>1.6.2</version>
    </dependency>
<!--Spark SQL 操作 Hive 的依赖-->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-hive_2.10</artifactId>
        <version>1.6.2</version>
    </dependency>
</dependencies>

1. 创建 DataFrame
   Spark 应用程序使用 SQLContext,可以通过 RDD、Hive表、JSON格式数据等数据源创建DataFrame。下面是基于 JSON 文件创建 DataFrame的示例。

{"name":"Michael"}
{"name":"Andy","age":30}
{"name":"Justin","age":19}

package com.carve.dataframe

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

object DataFrameApp {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("DataFrameApp")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)

    val df = sqlContext.read.json("/usr/local/hadoop/myData/people.json")

    df.show()
    sc.stop
  }

}

1. DataFrame 操作
   使用 printSchema 方法输出 DataFrame 的 Schema 信息

df.printSchema()

使用 select 方法来选择所需要的字段

df.select("name").show()

使用 select 方法选择所需要的字段，并为 age 字段加 1

df.select(df("name"),df("age")+1).show()

使用 filter 方法完成条件过滤

df.filter(df("age")>21).show()

使用 groupBy 方法进行分组，求分组后的总数

df.groupBy("age").count().show()

1. DataFrame SQL 操作
   Spark 可以使用 SQLContext 的 sql() 方法执行 SQL 查询操作，sql()方法返回的查询结果为 DataFrame 格式。代码如下：

// sql() 方法执行 SQL 查询操作
val df = sqlContext.read.json("/usr/local/hadoop/myData/people.json")
people.registerTempTable("people")
sqlContext.sql("select * from people where age>21")

1. RDD 与 DataFrame 互操作

• 第一种操作方式：使用放射获取 RDD 内的 Schema

// 将 RDD 转成 DataFrame
val people = sc.textFile("/usr/local/hadoop/myData/people.json").map(_.split(",")).map(p=>Person(p(0),p(1).trim.toInt)).toDF()

// 将DataFrame注册成临时表，后续可以直接使用 SQL 进行查询
people.registerTempTable("people")
val teenagers = sqlContext.sql("SELECT name,age FROM people WHERE age>=13 AND age <= 19")
teenagers.show()
// DataFrame 转成 RDD进行操作：根据索引号取值
teenagers.map(t=>"Name:"+t(0)).collect().foreach(println)
// DataFrame 转成 RDD进行操作：根据字段名称取值
teenagers.map(t=>"Name:"+t.getAs[String]("name")).collect().foreach(println)
// DataFrame 转成 RDD进行操作：一次性返回多列的值
teenagers.map(_.getValuesMap[Any](List("name","age"))).collect().foreach(println)

• 通过编程接口指定 Schema
  通过 Spark SQL 的接口创建 RDD 的 Schema，可以动态生成 Schema。

4、创建DataFrame

在 Spark SQL 中 SparkSession 是创建 DataFrame 和执行 SQL 的入口，创建 DataFrame 有三种方式：通过 Spark 的数据源进行创建；从一个存在的 RDD 进行转换；还可以从 Hive Table 进行查询返回。

1. 从 Spark 数据源进行创建

# 查看Spark支持创建文件的数据源格式
scala> spark.read.
csv format jdbc json load option options orc parquet schema 
table text textFile
# 在 spark 的 bin/data 目录中创建 user.json 文件
{"username":"zhangsan","age":20}
# 读取 json 文件创建 DataFrame
scala> val df = spark.read.json("/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint， username: string]
scala> df.show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+

1. 从 RDD 进行转换
   在 IDEA 中开发程序时，如果需要 RDD 与 DF 或者 DS 之间互相操作，那么需要引入 import spark.implicits._ 这里的 spark 不是 Scala 中的包名，而是创建的 sparkSession 对象的变量名称，所以必 须先创建 SparkSession 对象再导入。这里的 spark 对象不能使用 var 声明，因为 Scala 只支持 val 修饰的对象的引入。 spark-shell 中无需导入，自动完成此操作。

scala> val idRDD = sc.textFile("/id.txt")
scala> idRDD.toDF("id").show
+---+
| id|
+---+
| 1|
| 2|
| 3|
| 4|
+---+

实际开发中，一般通过样例将 RDD 转换为 DataFrame

scala> case class User(name:String, age:Int)
defined class User
scala> sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1, t._2)).toDF.show

val people = sc.parallelize(List("Tom,14",
"Jerry,15",
"Mike,18",
"Andy,20",
"Jake,19"))
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
val rdd = people.map(_.split(",")).map(p => Row(p(0),p(1).trim.toInt))    //将people RDD转换为ROW
val schema = StructType(Array(StructField("name",StringType, true) ,StructField("age",IntegerType, true) ) )      //创建Schema
val peopleDF = spark.sqlContext.createDataFrame(rdd,schema)    //将RDD转为DataFrame
peopleDF.printSchema()                  //输出Schema
peopleDF.show()        //输出全部

1. 从 Hive Table 进行查询返回

5、SQL 语法

SQL 语法风格是指我们查询数据的时候使用 SQL 语句来查询，这种风格的查询必须要 有临时视图或者全局视图来辅助

1. 读取 JSON 文件创建 DataFrame

scala> val df = spark.read.json("/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint， username: string]

1. 对 DataFrame 创建一个临时表

scala> df.createOrReplaceTempView("people")

1. 通过 SQL 语句实现查询全表

scala> val sqlDF = spark.sql("SELECT * FROM people")
sqlDF: org.apache.spark.sql.DataFrame = [age: bigint， name: string]

1. 结果展示

scala> sqlDF.show
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+

注意：普通临时表是 Session 范围内的，如果想应用范围内有效，可以使用全局临时表。使 用全局临时表时需要全路径访问，如：global_temp.people

1. 对 DataFrame 创建一个全局表

scala> df.createGlobalTempView("people")

1. 通过 SQL 语句实现查询全表

scala> spark.sql("select * from global_temp.people").show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+

scala> spark.newSession().sql("SELECT * FROM global_temp.people").show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
+---+--------+

6、DSL 语法

DataFrame 提供了一个特定领域语言(domain-specific language,DSL)去管理结构化的数据。可以在 Scala,Java,Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了

1. 创建一个 DataFrame

scala> val df = spark.read.json("/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

1. 查看 DataFrame 的 Schema 信息

scala> df.printSchema
root
 |-- age: long (nullable = true)
 |-- username: string (nullable = true)

1. 只查看 “username”列数据

scala> df.select("username").show()
+--------+
|username|
+--------+
|zhangsan|
+--------+

1. 查看“username”列数据以及“age+1”数据
   注意:涉及到运算的时候, 每列都必须使用$, 或者采用引号表达式：单引号+字段名

scala> df.select($"username",$"age"+1).show()
scala> df.select('username,'age+1).show()

1. 查看“age”大于“30”的数据

scala> df.filter($"age">30).show

1. 按照“age”分组，查看数据条数

scala> df.groupBy("age").count.show

7、DataFrame 转换为 RDD

DataFrame 其实就是对 RDD 的封装，所以可以直接获取内部的 RDD

scala> val df = sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1, t._2)).toDF
df: org.apache.spark.sql.DataFrame = [name: string, age: int]
scala> val rdd = df.rdd
rdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[46] 
at rdd at <console>:25
scala> val array = rdd.collect
array: Array[org.apache.spark.sql.Row] = Array([zhangsan,30], [lisi,40])

注意：此时得到的 RDD 存储类型为 Row

scala> array(0)
res28: org.apache.spark.sql.Row = [zhangsan,30]
scala> array(0)(0)
res29: Any = zhangsan
scala> array(0).getAs[String]("name")
res30: String = zhangsan