Spark SQL
DataFrame 的创建以及基本操作
DataFrame可以理解成关系型数据库中的表，它与 RDD 的差别在于 DataFrame 有 schema 信息
public class DataFrameCreate {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setMaster("local").setAppName("DataFrame");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
DataFrame dataFrame = sqlContext.read()
.json(Thread.currentThread().getContextClassLoader().getResource("student.txt").getPath());
// 打印 DataFrame
dataFrame.show();
// 打印 DataFrame 的 Schema 信息
dataFrame.printSchema();
// 查询某一列的数据
dataFrame.select(dataFrame.col("name")).show();
// 查询某几列的数据，并对列进行计算
dataFrame.select(dataFrame.col("name"), dataFrame.col("age").plus(1)).show();
// 根据某一列的值进行过滤
dataFrame.filter(dataFrame.col("age").gt(20)).show();
// 根据某一列进行分组，然后再进行聚合
dataFrame.groupBy(dataFrame.col("name")).count().show();
}
}
object DataFrameCreate extends App {
val conf = new SparkConf().setMaster("local").setAppName("DataFrame")
val sc = SparkContext.getOrCreate(conf)
// 创建 SQLContext
val sqlContext = SQLContext.getOrCreate(sc)
// 创建 DataFrame 对象
val dataFrame = sqlContext.read.json(Thread.currentThread().getContextClassLoader.getResource("student.txt").getPath)
dataFrame.show()
dataFrame.printSchema()
dataFrame.select(dataFrame.col("name")).show()
dataFrame.select(dataFrame.col("name"), dataFrame.col("age").plus(1)).show()
dataFrame.filter(dataFrame.col("age").gt(20)).show()
dataFrame.groupBy(dataFrame.col("name")).count().show()
}
RDD转换为DataFrame
使用反射的方式将RDD转换为DataFrame
/**
* Java语言实现
* 通过反射的方式将RDD转换为DataFrame
*/
public class RDD2DataFrameReflection {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrameReflection");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
// 读取文件生成RDD，然后将RDD中的line转换为Person类型
JavaRDD<Person> initRDD = sc.textFile(Thread.currentThread().getContextClassLoader().getResource("person.txt").getPath())
.map(line -> line.split(" "))
.map(array -> new Person(array[0], Integer.parseInt(array[1]), array[2]));
// 通过反射创建DataFrame，然后注册成一张临时表（person）
sqlContext.createDataFrame(initRDD, Person.class).registerTempTable("person");
// 执行sql查询，查询出年龄大于18岁的数据
DataFrame personDataFrame = sqlContext.sql("select name,age,sex from person where age > 18");
// 打印数据
personDataFrame.show();
// 将DataFrame再次转换成RDD,此时RDD中的数据类型为RDD，需要将其转换成Person类型。最后打印结果。
personDataFrame.javaRDD()
.map(row -> new Person(row.getString(0), row.getInt(1), row.getString(2)))
.collect()
.forEach(System.out::println);
}
/**
* 使用反射将RDD转换为DataFrame时，JavaBean的class必须使用public修饰，并且需要实现Serializable接口
*/
public static class Person implements Serializable {
private String name;
private Integer age;
private String sex;
public Person() {
}
public Person(String name, Integer age, String sex) {
this.name = name;
this.age = age;
this.sex = sex;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
public String getSex() {
return sex;
}
public void setSex(String sex) {
this.sex = sex;
}
@Override
public String toString() {
return "Person{" +
"name='" + name + '\'' +
", age=" + age +
", sex='" + sex + '\'' +
'}';
}
}
}
// 使用scala语言实现RDD和DataFrame之间的转换
object RDD2DataFrameReflection {
// 申明一个case class 作为bean对象
case class Person(name: String, age: Int, sex: String)
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrame")
val sc = SparkContext.getOrCreate(conf)
val sqlContext = SQLContext.getOrCreate(sc)
// scala中RDD转成DataFrame需要导入隐式转换的包
import sqlContext.implicits._
// 读取txt文件生成RDD，调用toDF转换成DataFrame，最后将DataFrame注册成一张临时表
sc.textFile(Thread.currentThread().getContextClassLoader.getResource("person.txt").getPath)
.map(_.split(" "))
.map(array => Person(array(0), array(1).trim.toInt, array(2)))
.toDF()
.registerTempTable("person")
// 直接调用DataFrame的rdd方法即可得到RDD，此时RDD中的数据类型为Row，需要转换成Person。最后进行打印
sqlContext.sql("select name,age,sex from person where age > 18")
.rdd
.map(row => Person(row.getString(0), row.getInt(1), row.getString(2)))
.collect
.foreach(println)
}
}
使用编码的方式动态创建元数据
// 使用Java语言实现RDD和DataFrame的互相转换（自定义元数据的方式）
public class RDD2DataFrameProgrammatically {
public static void main(String[] args) {
// 创建SparkContext和SQLContext
SparkConf conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrame");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
// 从文件中读取数据生成RDD，将RDD中的类型转换成Row类型
JavaRDD<Row> rdd = sc.textFile(Thread.currentThread().getContextClassLoader().getResource("person.txt").getPath())
.map(line -> line.split(" "))
.map(array -> RowFactory.create(array[0], Integer.parseInt(array[1]), array[2]));
// 构建元数据信息
List<StructField> structFields = Arrays.asList(
DataTypes.createStructField("name", DataTypes.StringType, true),
DataTypes.createStructField("age", DataTypes.IntegerType, true),
DataTypes.createStructField("sex", DataTypes.StringType, true)
);
StructType structType = DataTypes.createStructType(structFields);
// 将RDD转换成DataFrame，并将其注册成一张临时表
sqlContext.createDataFrame(rdd, structType).registerTempTable("person");
sqlContext.sql("select name,age,sex from person where age > 18")
.javaRDD()
.map(row -> new RDD2DataFrameReflection.Person(row.getString(0), row.getInt(1), row.getString(2)))
.collect()
.forEach(System.out::println);
}
}
// 使用scala语言实现RDD和DataFrame的互相转换（自定义元数据的方式）
object RDD2DataFrameProgrammatically {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrame")
val sc = SparkContext.getOrCreate(conf)
val sqlContext = SQLContext.getOrCreate(sc)
// 从文本文件中读取数据生成RDD
val rdd = sc.textFile(Thread.currentThread().getContextClassLoader.getResource("person.txt").getPath)
.map(_.split(" "))
.map(array => Row(array(0), array(1).toInt, array(2)))
// 构建元数据信息
val structType = StructType(Array(
StructField("name", StringType, true),
StructField("age", IntegerType, true),
StructField("sex", StringType, true)
))
// 将RDD转换成DataFrame并将其注册成一张临时表
sqlContext.createDataFrame(rdd, structType).registerTempTable("person")
// 查询出年龄大于18的用户信息，将其转成RDD再进行打印
sqlContext.sql("select name,age,sex from person where age > 18")
.rdd
.map(row => Person(row.getString(0), row.getAs[Integer](1), row.getString(2)))
.collect()
.foreach(println)
}
}
load && save 方法
// Java 语言演示 load 和 save 方法的具体使用
public static void loadAndSave() {
SparkConf conf = new SparkConf().setAppName("LoadAndSave").setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
// 不指定类型时 load 和 save 默认的格式为 parquet
sqlContext.read()
.load(Thread.currentThread().getContextClassLoader().getResource("student.parquet").getPath())
.write()
.save(Thread.currentThread().getContextClassLoader().getResource("student_copy.parquet").getPath());
// 手动指定 load 和 save 方法操作的文件类型
sqlContext.read()
.format("json")
.load(Thread.currentThread().getContextClassLoader().getResource("student.txt").getPath())
.show();
sqlContext.read()
.format("json")
.load(Thread.currentThread().getContextClassLoader().getResource("student.txt").getPath())
.write()
.format("json")
.save("file:///Users/garen/Documents/student_copy");
sqlContext.read()
.format("json")
.load(Thread.currentThread().getContextClassLoader().getResource("student.txt").getPath())
.write()
.format("json")
.mode(SaveMode.ErrorIfExists)
.save("file:///Users/garen/Documents/student_copy");
}
def loadAndSave(): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("LoadAndSave")
val sc = SparkContext.getOrCreate(conf)
val sqlContext = SQLContext.getOrCreate(sc)
sqlContext.read
.format("json")
.load(Thread.currentThread().getContextClassLoader.getResource("student.txt").getPath)
.show()
val dataFrame = sqlContext.read
.format("json")
.load(Thread.currentThread().getContextClassLoader.getResource("student.txt").getPath)
dataFrame.filter(dataFrame.col("age").gt(18))
.write
.format("json")
.mode(SaveMode.ErrorIfExists)
.save("file:///D:/demo")
}
parquet数据源操作
使用编程方式加载parquet数据源
// java读取parquet文件
public static void loadParquet() {
SparkConf conf = new SparkConf().setMaster("local").setAppName("parquet");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
sqlContext.read().parquet(Thread.currentThread().getContextClassLoader().getResource("/tarball/users.parquet").getPath()).registerTempTable("person");
sqlContext.sql("select name from person")
.javaRDD()
.map(row -> row.getString(0))
.collect()
.forEach(System.out::println);
}
// scala读取parquet文件
def parquet(): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("parquet")
val sc = SparkContext.getOrCreate(conf)
val sqlContext = SQLContext.getOrCreate(sc)
sqlContext.read
.parquet(Thread.currentThread.getContextClassLoader.getResource("tarball/users.parquet").getPath)
.registerTempTable("users")
sqlContext.sql("select * from users")
.rdd
.map(_.getString(0))
.collect
.foreach(println)
}
自动分区推断
// Java语言
public static void parquetPartion() {
SparkConf conf = new SparkConf().setMaster("local").setAppName("ParquetPartion");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
// 打印schema信息，会多出两个分区字段("gender","country")
// 分区列数据类型：支持numeric和string类型的自动推断，通过“spark.sql.sources.partitionColumnTypeInference.enabled”配置开启或关闭（默认开启），关闭后分区列全为string类型。
sqlContext.read()
.parquet("hdfs://spark1:9000/spark-study/users/gender=male/country=US/users.parquet")
.printSchema();
}
自动合并元数据
def mergeSchema(): Unit = {
val conf: SparkConf = new SparkConf().setMaster("local").setAppName("MergeSchema")
// 方法一：通过 conf 对象，设置 spark.sql.parquet.mergeSchema 为 true
// conf.set("spark.sql.parquet.mergeSchema", "true")
val sc: SparkContext = SparkContext.getOrCreate(conf)
val sqlContext: SQLContext = SQLContext.getOrCreate(sc)
val nameAndAgeArray: Array[(String, Int)] = Array(("Tom", 23), ("Garen", 21), ("Peter", 25))
val nameAndScoreArray: Array[(String, Int)] = Array(("Tome", 100), ("Garen", 98), ("Peter", 95))
import sqlContext.implicits._
sc.parallelize(nameAndAgeArray)
.toDF("name", "age")
.write
.format("parquet")
.mode(SaveMode.Append)
.save("hdfs://users/hadoop/persons")
sc.parallelize(nameAndScoreArray)
.toDF("name", "score")
.write
.format("parquet")
.mode(SaveMode.Append)
.save("hdfs://users/hadoop/persons")
// 方法二：通过 option 设置 MergeSchema 为 true
sqlContext.read.option("mergeSchema", "true")
.parquet("hdfs://users/hadoop/persons")
.printSchema()
sqlContext.read.option("mergeSchema", "true")
.parquet("data/persons")
.show()
}
/**
* root
* |-- name: string (nullable = true)
* |-- score: integer (nullable = true)
* |-- age: integer (nullable = true)
*
* +-----+-----+----+
* | name|score| age|
* +-----+-----+----+
* | Tom| null| 23|
* |Garen| null| 21|
* |Peter| null| 25|
* | Tome| 100|null|
* |Garen| 98|null|
* |Peter| 95|null|
* +-----+-----+----+
*/
Json 数据源操作
需求：查询分数大于80的学生信息以及成绩信息
public static void jsonOption() {
SparkConf conf = new SparkConf().setMaster("local").setAppName("JsonOption");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
// 方式一： 从 json 文件创建学生信息的 DataFrame，并注册成临时表
sqlContext.read().format("json")
.load(Thread.currentThread().getContextClassLoader().getResource("student.txt").getPath())
.registerTempTable("student_info");
JavaRDD<String> rdd = sc.parallelize(Arrays.asList(
"{\"id\":\"1001\",\"score\":100}",
"{\"id\":\"1002\",\"score\":79}",
"{\"id\":\"1003\",\"score\":98}"
));
// 方式二： 从 json 格式的 RDD 创建学生分数信息的 DataFrame，并注册成临时表
sqlContext.read().json(rdd).registerTempTable("student_score");
// 查询分数大于80的学生信息和成绩，并保存到 json 文件中
sqlContext.sql("select a.id,a.name,a.age,b.score from student_info a left join student_score b on a.id = b.id where b.score > 80")
.write()
.format("json")
.mode(SaveMode.Overwrite)
.save("data/student_score");
}
def jsonOption(): Unit = {
val conf: SparkConf = new SparkConf().setMaster("local").setAppName("JsonOption")
val sc: SparkContext = SparkContext.getOrCreate(conf)
val sqlContext = SQLContext.getOrCreate(sc)
// 从文件中读取json数据，并注册成临时表
sqlContext.read
.format("json")
.load(Thread.currentThread().getContextClassLoader.getResource("student.txt").getPath)
.registerTempTable("student_info")
// 从rdd中读取数据，并注册成临时表
val jsonArray: Array[String] = Array(
"{\"id\":\"1001\",\"score\":100}",
"{\"id\":\"1002\",\"score\":90}",
"{\"id\":\"1003\",\"score\":80}"
)
val rdd = sc.parallelize(jsonArray)
sqlContext.read.json(rdd).registerTempTable("student_score")
sqlContext.sql("select a.id,a.name,a.age,b.score from student_info a left join student_score b on a.id = b.id where b.score > 80")
.write
.format("json")
.mode(SaveMode.Overwrite)
.save("data/students_score")
}
Hive数据源操作
操作hive数据源完成查询成绩大于80的用户信息的功能。对hive数据源进行操作时需要将hive-site.xml和mysql的驱动包拷贝到spark的目录下。
public static void hiveOption() {
SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("HiveOption");
JavaSparkContext sc = new JavaSparkContext(conf);
HiveContext hiveContext = new HiveContext(sc);
// 如果student_info表存在则先删除
hiveContext.sql("DROP TABLE IF EXISTS hive.student_info").show();
// 创建studnet_info表结构
hiveContext.sql("CREATE TABLE IF NOT EXISTS hive.student_info (name STRING,age INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ' '").show();
// 加载数据到student_info表中
hiveContext.sql("LOAD DATA INPATH '/user/hadoop/test_data/student_info.txt' INTO TABLE student_info").show();
// 如果student_score表存在则先删除
hiveContext.sql("DROP TABLE IF EXISTS hive.student_score").show();
// 创建student_score表结构
hiveContext.sql("CREATE TABLE IF NOT EXISTS hive.student_score (name STRING,score INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ' '").show();
// 加载数据到student_score表中
hiveContext.sql("LOAD DATA INPATH '/user/hadoop/test_data/student_score.txt' INTO TABLE student_score").show();
// 查询出成绩大于80的学生信息并保存为hive的一张新表
hiveContext.sql("select a.name,a.age,b.score from student_info a left join student_score b on a.name = b.name where b.score > 80")
.saveAsTable("student_info_score");
// 通过hive中的表直接创建DataFrame
Row[] rows = hiveContext.table("student_info_score").collect();
Arrays.asList(rows).forEach(System.out::println);
}
JDBC 数据源操作
public static void jdbc() {
SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("JDBCOption");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = SQLContext.getOrCreate(sc.sc());
// 从student_info表中读取数据，并转换成 RDD
Map<String, String> options = new HashMap<>();
options.put("url", "jdbc:mysql://hadoop-cdh:3306/testdb");
options.put("user", "root");
options.put("password", "Slb930802.");
options.put("dbtable", "student_info");
DataFrame student_info_df = sqlContext.read().format("jdbc").options(options).load();
JavaPairRDD<String, Integer> student_info_rdd = student_info_df.javaRDD()
.mapToPair(row -> new Tuple2<String, Integer>(row.getString(0), row.getInt(1)));
// 从student_score表中读取数据，并转换成 RDD
options.put("dbtable", "student_score");
DataFrame student_score_df = sqlContext.read().format("jdbc").options(options).load();
JavaPairRDD<String, Integer> student_score_rdd = student_score_df.javaRDD()
.mapToPair(row -> new Tuple2<String, Integer>(row.getString(0), row.getInt(1)));
// 使用 join 算子连接两个 RDD，再筛选出分数大于80的学生信息
JavaRDD<Row> good_student_info_rdd = student_info_rdd.join(student_score_rdd)
.map(t -> RowFactory.create(t._1(), t._2()._1(), t._2()._2()))
.filter(row -> row.getInt(2) > 80);
// 通过自定义元数据的方式将 RDD 转换成 DataFrame 并打印
List<StructField> structFields = Arrays.asList(
DataTypes.createStructField("name", DataTypes.StringType, true),
DataTypes.createStructField("age", DataTypes.IntegerType, true),
DataTypes.createStructField("score", DataTypes.IntegerType, true)
);
StructType structType = DataTypes.createStructType(structFields);
sqlContext.createDataFrame(good_student_info_rdd, structType).show();
// 将 RDD 的数据保存到 mysql 数据库中
good_student_info_rdd.foreach(row -> {
Class.forName("com.mysql.jdbc.Driver");
Connection conn = null;
PreparedStatement ps = null;
try {
conn = DriverManager.getConnection("jdbc:mysql://hadoop-cdh:3306/testdb", "root", "Slb930802.");
ps = conn.prepareStatement("INSERT INTO good_student_info VALUES(?,?,?)");
ps.setString(1, row.getString(0));
ps.setInt(2, row.getInt(1));
ps.setInt(3, row.getInt(2));
ps.executeUpdate();
} catch (Exception e) {
e.printStackTrace();
} finally {
if (ps != null) ps.close();
if (conn != null) conn.close();
}
});
}
SparkSQL内置函数
统计UV
def uv(): Unit = {
val conf = new SparkConf().setMaster("local[4]").setAppName("uv")
val sc = SparkContext.getOrCreate(conf)
val sqlContext = SQLContext.getOrCreate(sc)
// 构造日志信息，并转换成DataFrame的格式
val logArray = Array(
"20180115,fpanpxc2gvjd2py0jd2254ng",
"20180115,fpanpxc2gvjd2py0jd2254ng",
"20180115,4xlt0txdgy220f3owkhxv4y1",
"20180116,kz5t0lvcjsbsqacjqwqyhfu4")
val logRDD = sc.parallelize(logArray)
.map(_.split(","))
.map(array => RowFactory.create(array(0), array(1)))
val structType = StructType(Array(
StructField("data", StringType, false),
StructField("sessionId", StringType, false)
))
val logDF = sqlContext.createDataFrame(logRDD, structType)
// 使用SparkSQL的内置函数对字段做聚合操作
// 使用SparkSQL的内置函数需要导入以下两个包
import sqlContext.implicits._
import org.apache.spark.sql.functions._
logDF.groupBy("data").agg(countDistinct('sessionId).as("uv")).show()
}
统计每日的总销售额
def totalSale(): Unit = {
val conf = new SparkConf().setMaster("local[4]").setAppName("TotalSale")
val sc = SparkContext.getOrCreate(conf)
val sqlContext = SQLContext.getOrCreate(sc)
// 构建DataFrame
val logArray = Array("20180115,34.3",
"20180115,35.6",
"20180116,99.9")
val logRDD = sc.parallelize(logArray)
.map(_.split(","))
.map(array => Row(array(0), array(1).toDouble))
val structType = StructType(Array(
StructField("data", StringType, false),
StructField("sale", DoubleType, false)
))
// 做分组聚合，统计每日交易额
import sqlContext.implicits._
import org.apache.spark.sql.functions._
sqlContext.createDataFrame(logRDD, structType)
.groupBy("data")
.agg('data, sum('sale).as("total_sale"))
.show()
}
SparkSQL窗口函数
row_number()窗口函数
select url,rate,row_number() over(partition by url order by rate desc) as r from window_test2;
