码迷，mamicode.com
首页 > 数据库 > 详细

大数据-sparkSQL

时间：2019-11-01 09:24:31      阅读：82      评论：0      收藏：0      [点我收藏+]

标签：user   public   uri   用户信息   ali   rri   l命令   行数据   ignore   

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

 

 

大数据-sparkSQL

标签：user   public   uri   用户信息   ali   rri   l命令   行数据   ignore   

原文地址：https://www.cnblogs.com/javaxiaobu/p/11775071.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！