大数据-sparkSQL

SparkSQL采用Spark on Hive模式，hive只负责数据存储，Spark负责对sql命令解析执行。

SparkSQL基于Dataset实现，Dataset是一个分布式数据容器，Dataset中同时存储原始数据和元数据(schema)

Dataset的底层封装了RDD，Row类型的RDD就是Dataset< Row >,DataFrame

Dataset数据源包括：json，JDBC，hive，parquet，hdfs，hbase，avro...

API

自带API

Dataset自带了一套api能够对数据进行操作，处理逻辑与sql处理逻辑相同。

//ds代表了一个Dataset<Row>,包括字段：age，name
//select name from table
ds.select(ds.col("name")).show();
//select name ,age+10 as addage from table
ds.select(ds.col("name"),ds.col("age").plus(10).alias("addage")).show();
//select name ,age from table where age>19
ds.select(ds.col("name"),ds.col("age")).where(ds.col("age").gt(19)).show();
//select age,count(*) from table group by age
ds.groupBy(ds.col("age")).count().show();

SQL的API

将Dataset转为临时表，在通过session对象执行sql，sql结果为Dataset类型

//将Dataset数据临时注册为临时表，指定表名。可以使用两种方法创建
ds.registerTempTable("table1");
ds.createOrReplaceTempView("table2");
//执行sql返回结果Dataset，sql可以同时使用多张临时表
Dataset<Row> result = sparkSession.sql("select * from table1");

Dataset方法

//打印元数据，以树形结构呈现
ds.printSchema();
​
//展示dataset的数据，默认显示二十条，可以指定显示条数。显示时数据列以ascii排序显示
ds.show();
​
//将dataset转为RDD<Row> 
JavaRDD<Row> rowRDD = ds.javaRDD();
//将JavaRDD<Row>解析为普通RDD
JavaRDD<String> map = rowRDD.map(new Function<Row, String>() {
private static final long serialVersionUID = 1L;
    @Override
    public String call(Row row) throws Exception {
        //方式1 指定字段名
        String id = row.getAs("id");
        String name = row.getAs("name");
        Integer age = row.getAs("age");
        //方式2 指定字段顺序。不常用。在json新构建的dataset会对字段按ASCII排序
        id = row.getString(1);
        name = row.getString(2);
        age = row.getInt(0);
        //返回结果
        return id + name + age;
    }
});
​

输出Dataset

Dataset的保存策略(SaveMode)包括：Overwrite(覆盖)，Append(追加)，ErrorIfExists(如果存在就报错)，Ignore(如果存在就忽略)

jdbc输出

Dataset<Row> dataset=sparkSession.sql("...");
//构建参数，存入用户名和用户密码
Properties userinfo = new Properties();
properties.setProperty("user", "root");
properties.setProperty("password", "root");
​
//指定写出模式，数据库连接，用户信息
result.writer().
    mode(SaveMode.Overwrite).
    jdbc( "jdbc:mysql://127.0.0.1:3306/spark",  "table", userinfo);

json输出

df.write().mode(SaveMode.Overwrite).format("json").save("data/parquet");
df.write().mode(SaveMode.Overwrite).json("data/parquet");

parquet输出

df.write().mode(SaveMode.Overwrite).format("parquet").save("data/parquet");
df.write().mode(SaveMode.Overwrite).parquet("data/parquet");

hive输出

在session中预先开启hive支持，在项目文件中导入hive，hdfs的3个xml

//对session执行hive语句，注意：需要指定表空间
sparkSession.sql("use database");
//移除原hive表
sparkSession.sql("DROP TABLE IF EXISTS table01");
//将dataset存入hive中       
df.write().mode(SaveMode.Overwrite).saveAsTable("table01");

Dataset创建

通过session创建和停止

//创建session对象
SparkSession sparkSession = SparkSession
                .builder()
                .appName("jsonrdd")
                .master("local")
                .getOrCreate();
/* 逻辑代码 */
​
//关闭session
sparkSession.stop();

基于json文件

• 基于json的构建，在json中已经存在了元数据，可以直接构造dataset

• 不支持嵌套json

//通过session对象读取json文件构建Dataset
Dataset<Row> ds = sparkSession.read().format("json").load("data/json");
//简写方式：Dataset<Row> ds =sparkSession.read.json("data/json");

//将Dataset转为临时表
ds.createOrReplaceTempView("table")

基于RDD< json>

RDD的每条数据都是一个json字符串，一个单行数据，自带了元数据信息

//通过上下文构建RDD
SparkContext sc = sparkSession.sparkContext();
JavaSparkContext jsc = new JavaSparkContext(sc);
JavaRDD<String> RDD = jsc.parallelize(Arrays.asList(
    "{"name":"zs","score":"100"}",
    "{"name":"sl","score":"200"}",
    "{"name":"ww","score":"300"}"
));
//通过RDD构建dataset
Dataset<Row> ds = sparkSession.read().json(RDD);

对象list创建

//构建存储对象的容器
Person person = new Person();
person.setId("1");
person.setAge(18);
person.setName("zs");
Person person2 = new Person();
person2.setId("2");
person2.setAge(20);
person2.setName("ls");
List<Person> people = Arrays.asList(person, person2);

//对bean类进行转码
Encoder<Person> personEncoder = Encoders.bean(Person.class);

//获取person类型的dataset，内部自带了
Dataset<Person> dataset = sparkSession.createDataset(people, personEncoder);

dataset.registerTempTable("person");

反射创建

不建议使用，修改字段较为麻烦

• 需要预先创建数据的javabean对象，实现序列化

• 读取数据构建RDD< String>，通过map算子转为RDD< Bean对象 >

• createDataFrame传入RDD与bean.class

//通过上下文构建RDD
SparkContext sc = sparkSession.sparkContext();
JavaSparkContext jsc = new JavaSparkContext(sc);
JavaRDD<String> lineRDD = jsc.textFile("/person.txt");
//将RDD的范型处理为bean类型
JavaRDD<Person> RDD = lineRDD.map(new Function<String, Person>() {
    //指定序列化版本
    private static final long serialVersionUID = 1L;
    @Override
    public Person call(String line) throws Exception {
        Person p = new Person();
        p.setId(line.split(",")[0]);
        p.setName(line.split(",")[1]);
        p.setAge(Integer.valueOf(line.split(",")[2]));
        return p;
    }
});
//将RDD转为Dataset，传入RDD和反射类
Dataset<Row> dataFrame = sparkSession.createDataFrame(RDD,Person.class);

bean对象

public class Person implements Serializable {
	/*指定序列化版本*/
	private static final long serialVersionUID = 1L;
	private String id ;
	private String name;
	private Integer age;
    //get+set
	public String getId() { return id; } public void setId(String id) {this.id = id; } 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; }
}

动态创建

• 读取数据构建RDD< String>

• 通过map算子将String转为Row，使用RowFactory的create方法存入数据

• 构建元数据对象schema，构建方法中传入元数据封装容器，内部存了每个字段的属性

• 通过RDD与schema构建出Dataset

• 对象中传入每个字段的属性

//通过上下文构建RDD
SparkContext sc = sparkSession.sparkContext();
JavaSparkContext jsc = new JavaSparkContext(sc);
JavaRDD<String> lineRDD = jsc.textFile("/person.txt");

//将一行数据转为Row类型，使用RowFactory的create方法
JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() {
    private static final long serialVersionUID = 1L;
    @Override
    public Row call(String line) throws Exception {
        //使用RowFactory.create构建row对象
        return RowFactory.create(
                line.split(",")[0],
                line.split(",")[1],
                Integer.valueOf(line.split(",")[2])
            );
    }
});

//构建字段属性容器,可以通过数据库查询后填充容器，实现动态调整
List<StructField> list = Arrays.asList(
    //分别构建每个字段，参数：字段名，数据类型，是否允许空值
    DataTypes.createStructField("id", DataTypes.StringType, true),
    DataTypes.createStructField("name", DataTypes.StringType, true),
    DataTypes.createStructField("age", DataTypes.IntegerType, true)
);
//构建schema对象
StructType schema = sparkSession.createStructType(list);

//构建Dataset
sparkSession.createDataFrame(rowRDD,schema)

JDBC创建

/* 方式1：创建参数容器传入 */
Map<String, String> options = new HashMap<String, String>();
options.put("url", "jdbc:mysql://127.0.0.1:3306/spark");
options.put("driver", "com.mysql.jdbc.Driver");
options.put("user", "root");
options.put("password", "root");
options.put("dbtable", "person");
//加载参数，传入参数
Dataset<Row> result = sparkSession.read().format("jdbc").options(options).load();

/* 方式2：依次传入参数 */
//创建读取器
DataFrameReader reader = sparkSession.read().format("jdbc");
//依次载入数据
reader.option("url", "jdbc:mysql://127.0.0.1:3306/spark");
reader.option("driver", "com.mysql.jdbc.Driver");
reader.option("user", "root");
reader.option("password", "root");
reader.option("dbtable", "score");
//加载数据，构建dataset
Dataset<Row> result = reader.load();

parquet创建

parque格式的文件中，存储了数据结构，因此可以直接构建dataset

//数据载入
Dataset load =sparkSession.read().format(parquet).load("data/parquet");
Dataset load = sparkSession.read().parquet("data/parquet");

Hive创建

• 在session中启动hive支持

• 项目中导入hive-site.xml，hdfs-site.xml，core-site.xml配置文件

• 需要hive中提前启动metastore服务

//开启hive支持，在项目文件中导入hive，hdfs的3个xml
SparkSession sparkSession = SparkSession
	.builder()
    .appName("hive")
    .enableHiveSupport()
    .getOrCreate();
//对session执行hive语句，注意：需要指定表空间
sparkSession.sql("use database");
    
//将hive_sql结果保存在为dataset
Dataset<Row> goodStudentsDF = sparkSession.sql("xxx");


//Dataset<Row> df = hiveContext.table("student_infos");

hive-site.xml的示例：

<configuration>  
<property>  
  <name>hive.metastore.uris</name>  
  <value>thrift://192.168.78.103:9083</value>  
</property>
</configuration>  

序列化

• 序列化与反序列化版本需要一致 private static final long serialVersionUID = 1L

• 子类实现序列化接口，父类未实现序列化接口，则子类中的父类字段无法序列化

  若父类实现了序列化，子类所有字段也能序列化

• 只序列化成员变量，不序列化静态变量。反序列化时，静态变量值返回原先值

• transuent修改的变量不序列化

结论：Java的序列化机制是通过在运行时判断类的serialVersionUID来验证版本一致性的。在进行反序列化时，JVM会把传来的字节流中的serialVersionUID与本地相应实体（类）的serialVersionUID进行比较，如果相同就认为是一致的，可以进行反序列化，否则就会出现序列化版本不一致的异常。当实现java.io.Serializable接口的实体（类）没有显式地定义一个名为serialVersionUID，类型为long的变量时，Java序列化机制会根据编译的class自动生成一个serialVersionUID作序列化版本比较用，这种情况下，只有同一次编译生成的class才会生成相同的serialVersionUID 。如果我们不希望通过编译来强制划分软件版本，即实现序列化接口的实体能够兼容先前版本，未作更改的类，就需要显式地定义一个名为serialVersionUID，类型为long的变量，不修改这个变量值的序列化实体都可以相互进行串行化和反串行化。

自定义函数

UDF

• 通过sparkSession注册UDF，实现UDF函数（UDF0-UDF22）

• 注册后的UDF在当前的session中有效

//获取session
SparkSession sparkSession = SparkSession
        .builder()
        .appName("udf")
        .master("local")
        .enableHiveSupport()
        .getOrCreate();
//通过session创建UDF，指定函数名，函数逻辑，返回值类型
sparkSession.udf().register(
        "add",
        new UDF2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }, DataTypes.IntegerType);
sparkSession.sql("select add(a,b) from table")

UDAF

可以理解为自定义组函数

通过sparkSession注册函数，实现UserDefinedAggregateFunction类重写以下方法

• initialize：初始化中间结果缓存，每个分区的每个组初始化一次，总分区的每个组初始化一次

• upadte：处理数据，合并一个分区中各组的数据，入参为每个组的缓存和当前记录

• merge：将各分区的中间结果进行合并

• evaluate：指定最终结果，从最后一个中间结果中取出,数据类型自己指定

• dataType：指定结果字段的数据类型

• inputSchema：指定每个输入字段的的列名，数据类型，是否为空。可以指定多个字段

• bufferSchema：定义buffer中每个字段的属性（列名，数据类型，是否为空），可以指定多个字段

• deterministic：确保一致性，一般使用true

//通过session创建UDF
sparkSession.udf().register(
        "add",
        new UDF2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        }, DataTypes.IntegerType);


sparkSession.udf().register(
        "udaf", new UserDefinedAggregateFunction() {

            //初始化中间结果缓存，每个分区的每个组初始化一次，总分区的每个组初始化一次
            @Override
            public void initialize(MutableAggregationBuffer buffer) {
                //内部可以存储多个数据
                buffer.update(0,0);
            }

            //处理数据,入参为每个组的缓存和当前记录
            @Override
            public void update(MutableAggregationBuffer buffer, Row input) {

                //数据封装在row中，从row中取值字段的值进行逻辑处理
                Object fieldName = input.getAs("fieldName");

                //对缓存中的数据更新
                buffer.update(0,buffer.getInt(0)+1);
            }

            //将各分区的中间结果进行合并
            //参数1：已经合并分区的中间结果或初始的中间结果，其初始化也使用initialize
            //参数2：当前分区的中间结果
            @Override
            public void merge(MutableAggregationBuffer buffer,  Row newBuffer) {
                //获取数据值
                Integer newData = newBuffer.getAs(0);
                Integer beforeData = buffer.getAs(0);
                //更新合并的中间结果
                buffer.update(0,beforeData+newData);
            }

            //指定最终结果，从最后一个中间结果中取出,数据类型自己指定
            @Override
            public Integer evaluate(Row buffer) {
                return buffer.getAs(0);
            }

            //指定每个输入字段的的列名，数据类型，是否为空。可以指定多个字段。
            @Override
            public StructType inputSchema() {
                return DataTypes.createStructType(
                        Arrays.asList(
                                DataTypes.createStructField(
                                        "name",
                                        DataTypes.StringType,
                                        true)
                        )
                );
            }

            //定义buffer中每个字段的属性（列名，数据类型，是否为空），可以指定多个字段
            @Override
            public StructType bufferSchema() {
                return DataTypes.createStructType(
                        Arrays.asList(
                                DataTypes.createStructField(
                                        "count",
                                        DataTypes.IntegerType,
                                        false),
                                DataTypes.createStructField(
                                        "name",
                                        DataTypes.StringType,
                                        true)
                        )
                );
            }

            //指定结果字段的数据类型
            @Override
            public DataType dataType(){return DataTypes.StringType;}

            //确保一致性，一般使用true
            @Override
            public boolean deterministic() { return true; }
        }
);

//上述UDAF模拟了count的功能
sparkSession.sql("select add(name) from table group by name ");

 

开窗函数

用于分组排序取最高值，适用于hive，mysql，oracle

row_number() over (partitin by xxx order by yyy) as sss

数据根据xxx列分组，再根据yyy排序，排序结果作为sss列的数据，通过对sss列行过滤，取出最大或最小的若干个值

SELECT
	t.id,
	t.uname,
	t.money
FROM
	(
		SELECT
			id,
			uname,
			money,
			row_number () over (
				PARTITION BY uname
				ORDER BY
					money DESC
			) rank
		FROM
			table
	) t
WHERE
	t.rank <= 3