1 Spark SQL

编程方式：（1）SQL;(2) DataFrame API

scala> case class Customer(id:Int,name:String,age:Int)
defined class Customerscala> val arr = Array("1,Mike,20","2,Mary,19","3,Jerry,23")
arr: Array[String] = Array(1,Mike,20, 2,Mary,19, 3,Jerry,23)scala> val rdd1 = sc.parallelize(arr)
rdd1: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[1] at parallelize at <console>:26scala> rdd1.collect
res1: Array[String] = Array(1,Mike,20, 2,Mary,19, 3,Jerry,23)scala> :paste
// Entering paste mode (ctrl-D to finish)rdd1.map(e=>{
val arr = e.split(",")
Customer(arr(0).toInt,arr(1),arr(2).toInt)
})// Exiting paste mode, now interpreting.res2: org.apache.spark.rdd.RDD[Customer] = MapPartitionsRDD[2] at map at <console>:31scala> val rdd2 = res2
rdd2: org.apache.spark.rdd.RDD[Customer] = MapPartitionsRDD[2] at map at <console>:31scala> rdd2.collect
res3: Array[Customer] = Array(Customer(1,Mike,20), Customer(2,Mary,19), Customer(3,Jerry,23))scala> val df = spark.createDataFrame(rdd2)
18/12/28 18:38:10 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
df: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]scala> df.printSchema
root|-- id: integer (nullable = false)|-- name: string (nullable = true)|-- age: integer (nullable = false)scala> df.show
+---+-----+---+
| id| name|age|
+---+-----+---+
|  1| Mike| 20|
|  2| Mary| 19|
|  3|Jerry| 23|
+---+-----+---+

scala> df.createTempView("customer")scala> val df2 = spark.sql("select * from customer")
df2: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]scala> df2.show
+---+-----+---+
| id| name|age|
+---+-----+---+
|  1| Mike| 20|
|  2| Mary| 19|
|  3|Jerry| 23|
+---+-----+---+scala> val df2 = spark.sql("select * from customer where id <2")
df2: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]scala> df2.show
+---+----+---+
| id|name|age|
+---+----+---+
|  1|Mike| 20|
+---+----+---+

scala> val df1 = spark.sql("select * from customer where id < 2")
df1: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]scala> df1.show
+---+----+---+
| id|name|age|
+---+----+---+
|  1|Mike| 20|
+---+----+---+scala> val df2 = spark.sql("select * from customer where id > 2")
df2: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]scala> df2.show
+---+-----+---+
| id| name|age|
+---+-----+---+
|  3|Jerry| 23|
+---+-----+---+// union => 纵向连接
scala> val dff = spark.sql("select * from c1 union select * from c2")
dff: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]scala> dff.show
+---+-----+---+
| id| name|age|
+---+-----+---+
|  3|Jerry| 23|
|  1| Mike| 20|
+---+-----+---+

2 Spark SQL 读取 json 文件

Dataset<Row> === DataFrame,类似于table的操作
SparkSession.read().json()
SparkSession.write().json()

2.1 Spark SQL Java 版本

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class SQLJava {public static void main(String[] args) {SparkSession session = SparkSession.builder().appName("SQLJava").config("spark.master", "local[2]").getOrCreate();Dataset<Row> df = session.read().json("d:/json.json");df.createOrReplaceTempView("stu");df = session.sql("select * from stu");df.show();}
}

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;import java.util.function.Consumer;public class SQLJava {public static void main(String[] args) {SparkSession session = SparkSession.builder().appName("SQLJava").config("spark.master", "local[2]").getOrCreate();Dataset<Row> df1 = session.read().json("d:/json.json");df1.createOrReplaceTempView("stu");df1 = session.sql("select * from stu");df1.show();Dataset<Row> df2 = session.sql("select * from stu where age > 20");df2.show();System.out.println("=============================");//聚合查询Dataset<Row> dfCount = session.sql("select count(*) from stu");dfCount.show();/** DataFrame 转换为 RDD* */JavaRDD<Row> rdd = df1.toJavaRDD();rdd.collect().forEach(new Consumer<Row>() {public void accept(Row row) {Long id = row.getAs("id");String name = row.getAs("name");Long age = row.getAs("age");System.out.println(id + "-" + name + "-" + age);}});}
}

大数据实时计算Spark学习笔记（9）—— Spar SQL(1) 读取 json 文件相关推荐

.NET 大数据实时计算--学习笔记
摘要纯 .Net 自研大数据实时计算平台,在中通快递服务数百亿包裹,处理数据万亿计!将分享大数据如何落地以及设计思路,技术重难点. 目录背景介绍计算平台架构项目实战背景介绍计算平台架构分 ...
1. 大数据实时计算介绍
Spark Streaming,其实就是一种Spark提供的,对于大数据,进行实时计算的一种框架.它的底层,其实,也是基于我们之前讲解的Spark Core的.基本的计算模型,还是基于内存的大数据实时 ...
【大数据实时计算框架】Storm框架
一.大数据实时计算框架 1.什么是实时计算?流式计算? (一)什么是Storm? Storm为分布式实时计算提供了一组通用原语,可被用于"流处理"之中,实时处理消息并更新数据库.这 ...
大数据实时计算工程师/Hadoop工程师/数据分析师职业路线图
大数据实时计算工程师/Hadoop工程师/数据分析师职业路线图 http://edu.51cto.com/roadmap/view/id-29.html http://my.oschina.net/i ...
接近淘宝 80%的大数据实时计算平台，从0搭建的经验和坑
上周一,来自武汉的直播平台斗鱼TV宣布C轮融资,腾讯领投的 15 亿人民币,距其获得 B 轮1亿美元不到半年,也是大写的牛逼. 但小寻更关心他们的大数据架构,作为一个在 2 年多时间里崛起的公司,其流 ...
Flink大数据实时计算系列-案例初体验：HotPages
Flink大数据实时计算系列-案例初体验:HotPages 目录 HotPages代码输入日志运行结果 HotPages代码 /*** Copyright (c) 2018-2028 尚硅谷 Al ...
Flink大数据实时计算系列-Flink的Operator Chains的优化机制
Flink大数据实时计算系列-Flink的Operator Chains的优化机制目录 Flink改变并行度并行度改为3 并行度改为2 Flink Operator Chains Flink gr ...
Flink大数据实时计算系列-Flink写出多个parquet小文件处理方法、Presto的介绍与使用场景
Flink大数据实时计算系列-Flink写出多个parquet小文件处理方法.Presto的介绍与使用场景 Presto的安装与使用目录 Flink写出多个parquet小文件处理方法 Presto ...
Flink大数据实时计算系列-列式存储parquet文件格式介绍、Flink进行rowformat格式文件保存
Flink大数据实时计算系列-列式存储parquet文件格式介绍 Flink进行rowformat格式文件保存列式存储parquet文件格式介绍

大数据实时计算Spark学习笔记（9）—— Spar SQL(1) 读取 json 文件

1 Spark SQL

2 Spark SQL 读取 json 文件

2.1 Spark SQL Java 版本

大数据实时计算Spark学习笔记（9）—— Spar SQL(1) 读取 json 文件相关推荐

最新文章

热门文章