spark-sql建表语句限制_SparkSQL

SparkSQL介绍

Hive是Shark的前身，Shark是SparkSQL的前身,SparkSQL产生的根本原因是其完全脱离了Hive的限制。

SparkSQL支持查询原生的RDD。 RDD是Spark平台的核心概念，是Spark能够高效的处理大数据的各种场景的基础。

能够在scala中写SQL语句。支持简单的SQL语法检查，能够在Scala中写Hive语句访问Hive数据，并将结果取回作为RDD使用。

Spark on Hive和Hive on Spark

Spark on Hive： Hive只作为储存角色，Spark负责sql解析优化，执行。

Hive on Spark：Hive即作为存储又负责sql的解析优化，Spark负责执行。

Dataset与DataFrame

Dataset也是一个分布式数据容器。与RDD类似，然而Dataset更像传统数据库的二维表格，除了数据以外，还掌握数据的结构信息，即schema。同时，与Hive类似，Dataset也支持嵌套数据类型（struct、array和map）。从API易用性的角度上 看， Dataset API提供的是一套高层的关系操作，比函数式的RDD API要更加友好，门槛更低。

Dataset的底层封装的是RDD，当RDD的泛型是Row类型的时候，我们也可以称它为DataFrame。即Dataset<Row> = DataFrame

SparkSQL底层架构

首先拿到sql后解析一批未被解决的逻辑计划，再经过分析得到分析后的逻辑计划，再经过一批优化规则转换成一批最佳优化的逻辑计划，再经过SparkPlanner的策略转化成一批物理计划，随后经过消费模型转换成一个个的Spark任务执行。

谓词下推（predicate Pushdown）

创建Dataset的几种方式

读取json格式的文件创建Dataset

注意：

• json文件中的json数据不能嵌套json格式数据。
• Dataset是一个一个Row类型的RDD，ds.rdd()/ds.javaRdd()。
• 可以两种方式读取json格式的文件。
• df.show()默认显示前20行数据。
• Dataset原生API可以操作Dataset（不方便）。
• 注册成临时表时，表中的列默认按ascii顺序显示列。

java: SparkSession sparkSession = SparkSession.builder().appName("jsonfile").master("local").getOrCreate(); /*** Dataset的底层是一个一个的RDD RDD的泛型是Row类型。* 以下两种方式都可以读取json格式的文件*/ Dataset<Row> ds = sparkSession.read().format("json").load("data/json"); // Dataset<Row> ds = sparkSession.read().json("data/json"); // ds.show();/*** Dataset转换成RDD*/ JavaRDD<Row> javaRDD = ds.javaRDD(); /*** 显示 Dataset中的内容，默认显示前20行。如果现实多行要指定多少行show(行数)* 注意：当有多个列时，显示的列先后顺序是按列的ascii码先后显示。*/ // ds.show(); /*** 树形的形式显示schema信息*/ds.printSchema();/*** Dataset自带的API 操作Dataset*///select name from table// ds.select("name").show();//select name age+10 as addage from tableds.select(ds.col("name"),ds.col("age").plus(10).alias("addage")).show();//select name ,age from table where age>19ds.select(ds.col("name"),ds.col("age")).where(ds.col("age").gt(19)).show();//select count(*) from table group by ageds.groupBy(ds.col("age")).count().show();/*** 将Dataset注册成临时的一张表，这张表临时注册到内存中，是逻辑上的表，不会雾化到磁盘*/ds.createOrReplaceTempView("jtable");Dataset<Row> result= sparkSession.sql("select age,count(*) as gg from jtable group by age");result.show(); sparkSession.stop();

通过json格式的RDD创建Dataset

java: SparkSession sparkSession = SparkSession.builder().appName("jsonrdd").master("local").getOrCreate(); SparkContext sc = sparkSession.sparkContext(); JavaSparkContext jsc = new JavaSparkContext(sc);JavaRDD<String> nameRDD = sc.parallelize(Arrays.asList("{"name":"zhangsan","age":"18"}","{"name":"lisi","age":"19"}","{"name":"wangwu","age":"20"}" )); JavaRDD<String> scoreRDD = sc.parallelize(Arrays.asList( "{"name":"zhangsan","score":"100"}", "{"name":"lisi","score":"200"}", "{"name":"wangwu","score":"300"}" ));Dataset<Row> nameds = sparkSession.read().json(nameRDD); Dataset<Row> scoreds = sparkSession.read().json(scoreRDD); nameds.createOrReplaceTempView("nameTable"); scoreds.createOrReplaceTempView("scoreTable"); Dataset<Row> result =sparkSession.sql("select nameTable.name,nameTable.age,scoreTable.score "+ "from nameTable join scoreTable "+ "on nameTable.name = scoreTable.name"); result.show(); sparkSession.stop();非json格式的RDD创建Dataset 通过反射的方式将非json格式的RDD转换成Dataset 自定义类要可序列化 自定义类的访问级别是Public RDD转成Dataset后会根据映射将字段按Assci码排序 将Dataset转换成RDD时获取字段两种方式,一种是ds.getInt(0)下标获取（不推荐使用）， 另一种是ds.getAs(“列名”)获取（推荐使用） /** * 注意： * 1.自定义类必须是可序列化的 * 2.自定义类访问级别必须是Public * 3.RDD转成Dataset会把自定义类中字段的名称按assci码排序 */ SparkSession sparkSession = SparkSession.builder().appName("reflect").master("local").getOrCreate();SparkContext sc = sparkSession.sparkContext();JavaSparkContext jsc = new JavaSparkContext(sc);JavaRDD<String> lineRDD = jsc.textFile("data/person.txt");JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() {private static final long serialVersionUID = 1L;@Overridepublic 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;} }); /** * 传入进去Person.class的时候，sqlContext是通过反射的方式创建DataFrame * 在底层通过反射的方式获得Person的所有field，结合RDD本身，就生成了DataFrame */ Dataset<Row> dataFrame = sparkSession.createDataFrame(personRDD, Person.class);dataFrame.show();dataFrame.printSchema();dataFrame.registerTempTable("person"); Dataset sql = sparkSession.sql("select name,id,age from person where id = 2");sql.show();/** * 将Dataset转成JavaRDD * 注意： * 1.可以使用row.getInt(0),row.getString(1)...通过下标获取返回Row类型的数据， 但是要注意列顺序问题---不常用 * 2.可以使用row.getAs("列名")来获取对应的列值。 * */ JavaRDD<Row> javaRDD = df.javaRDD(); JavaRDD<Person> map = javaRDD.map(new Function<Row, Person>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Person call(Row row) throws Exception {Person p = new Person();//p.setId(row.getString(1));//p.setName(row.getString(2));//p.setAge(row.getInt(0));p.setId((String)row.getAs("id"));p.setName((String)row.getAs("name"));p.setAge((Integer)row.getAs("age"));return p;} }); map.foreach(x-> System.out.println(x));sc.stop();

动态创建Schema将非json格式的RDD转换成Dataset

java： SparkSession sparkSession = SparkSession.builder().appName("schema").master("local").getOrCreate();SparkContext sc = sparkSession.sparkContext(); JavaSparkContext jsc = new JavaSparkContext(sc);JavaRDD<String> lineRDD = jsc.textFile("data/person.txt"); JavaRDD<String> lineRDD = sc.textFile("./sparksql/person.txt"); /*** 转换成Row类型的RDD*/ JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Row call(String s) throws Exception {return RowFactory.create(String.valueOf(s.split(",")[0]),String.valueOf(s.split(",")[1]),Integer.valueOf(s.split(",")[2]));} }); /*** 动态构建DataFrame中的元数据，一般来说这里的字段可以来源自字符串，也可以来源于外部数据库*/ List<StructField> asList =Arrays.asList(DataTypes.createStructField("id", DataTypes.StringType, true),DataTypes.createStructField("name", DataTypes.StringType, true),DataTypes.createStructField("age", DataTypes.IntegerType, true) );StructType schema = DataTypes.createStructType(asList); Dataset<Row> df = sparkSession.createDataFrame(rowRDD, schema); df.show(); sc.stop();

读取parquet文件创建Dataset

注意：

• 可以将Dataset存储成parquet文件。保存成parquet文件的方式有两种

df.write().mode(SaveMode.Overwrite)format("parquet")

.save("./sparksql/parquet");

df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");

• SaveMode指定文件保存时的模式。

Overwrite：覆盖

Append：追加

ErrorIfExists：如果存在就报错

Ignore：如果存在就忽略

java:

SparkSession sparkSession = SparkSession.builder().appName("parquet").master("local").getOrCreate();Dataset<Row> df = sparkSession.read().json("data/json"); /*** 将Dataset保存成parquet文件，SaveMode指定存储文件时的保存模式* 保存成parquet文件有以下两种方式：*/ df.write().mode(SaveMode.Overwrite).format("parquet").save("./data/parquet"); df.write().mode(SaveMode.Overwrite).parquet("./data/parquet"); df.show(); /*** 加载parquet文件成DataFrame * 加载parquet文件有以下两种方式： */Dataset load = sparksession.read().format("parquet").load("./data /parquet");load = sparksession.read().parquet("./data /parquet"); load.show();sc.stop();

读取JDBC中的数据创建Dataset(MySql为例)

两种方式创建Dataset

java:

SparkSession sparkSession = SparkSession.builder().appName("mysql").master("local").getOrCreate();/*** 第一种方式读取MySql数据库表，加载为DataFrame*/ Map<String, String> options = new HashMap<String,String>(); options.put("url", "jdbc:mysql://192.168.179.4:3306/spark"); options.put("driver", "com.mysql.jdbc.Driver"); options.put("user", "root"); options.put("password", "123456"); options.put("dbtable", "person"); Dataset<Row> person = sparkSession.read().format("jdbc").options(options).load(); person.show(); person.createOrReplaceTempView("person"); /*** 第二种方式读取MySql数据表加载为Dataset*/ DataFrameReader reader = sparkSession.read().format("jdbc"); reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark"); reader.option("driver", "com.mysql.jdbc.Driver"); reader.option("user", "root"); reader.option("password", "123456"); reader.option("dbtable", "score"); Dataset<Row> score = reader.load(); score.show(); score.createOrReplaceTempView("score"); Dataset result = sparksession.sql("select person.id,person.name,score.score from person,score where person.name = score.name"); result.show(); /*** 将Dataset结果保存到Mysql中*/ Properties properties = new Properties(); properties.setProperty("user", "root"); properties.setProperty("password", "root"); result.write().mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties);sc.stop();

读取Hive中的数据加载成Dataset

java:

SparkSession sparkSession = SparkSession.builder().master("local").appName("hvie")//开启hive的支持，接下来就可以操作hive表了// 前提需要是需要开启hive metastore 服务.enableHiveSupport().getOrCreate();sparkSession.sql("USE spark");sparkSession.sql("DROP TABLE IF EXISTS student_infos"); //在hive中创建student_infos表sparkSession.sql("CREATE TABLE IF NOT EXISTS student_infos (name STRING,age INT) row format delimited fields terminated by 't' ");sparkSession.sql("load data local inpath '/root/student_infos' into table student_infos");//注意：此种方式，程序需要能读取到数据（如/root/student_infos），同时也要能读取到 metastore服务的配置信息。sparkSession.sql("DROP TABLE IF EXISTS student_scores"); sparkSession.sql("CREATE TABLE IF NOT EXISTS student_scores (name STRING, score INT) row format delimited fields terminated by 't'");sparkSession.sql("LOAD DATA "+ "LOCAL INPATH '/root/student_scores'"+ "INTO TABLE student_scores");// Dataset<Row> df = hiveContext.table("student_infos");//读取Hive表加载Dataset方式/*** 查询表生成Dataset*/Dataset<Row> goodStudentsDF = sparkSession.sql("SELECT si.name, si.age, ss.score "+ "FROM student_infos si "+ "JOIN student_scores ss "+ "ON si.name=ss.name "+ "WHERE ss.score>=80");goodStudentsDF.registerTempTable("goodstudent");Dataset<Row> result = sparkSession.sql("select * from goodstudent");result.show();/*** 将结果保存到hive表 good_student_infos*/sparkSession.sql("DROP TABLE IF EXISTS good_student_infos"); goodStudentsDF.write().mode(SaveMode.Overwrite).saveAsTable("good_student_infos");sparkSession.stop();

Spark On Hive的配置

在Spark客户端配置spark On hive

在Spark客户端安装包下spark-2.2.1/conf中创建文件hive-site.xml：

配置hive的metastore路径

<configuration>

<property>

<name>hive.metastore.uris</name>

<value>thrift://node1:9083</value>

</property>

</configuration>

启动Hive的metastore服务

hive --service metastore &

启动zookeeper集群，启动HDFS集群。

启动SparkShell 读取Hive中的表总数，对比hive中查询同一表查询总数测试时间。

./spark-shell

--master spark://node1:7077

spark.sql("select * from day_table").show;

• 注意：

如果使用Spark on Hive 查询数据时，出现错误：

找不到HDFS集群路径，要在客户端机器conf/spark-env.sh中设置HDFS的路径：

序列化问题。

储存DataSet

将DataSet存储为parquet文件。

将DataSet存储到JDBC数据库。

将DataSet存储到Hive表。

自定义函数UDF和UDAF

UDF:用户自定义函数。

java:

SparkSession sparkSession = SparkSession.builder().appName("udf").master("local").getOrCreate();JavaSparkContext sc = new JavaSparkContext(sparkSession.sparkContext()); JavaRDD<String> parallelize = sc.parallelize(Arrays.asList("zhangsan","lisi","wangwu"));JavaRDD<Row> rowRDD = parallelize.map(new Function<String, Row>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Row call(String s) throws Exception {return RowFactory.create(s);} });/*** 动态创建Schema方式加载DF*/ List<StructField> fields = new ArrayList<StructField>(); fields.add(DataTypes.createStructField("name", DataTypes.StringType,true)); StructType schema = DataTypes.createStructType(fields);Dataset<Row> df = sparkSession.createDataFrame(rowRDD,schema);df.registerTempTable("user");/*** 根据UDF函数参数的个数来决定是实现哪一个UDF UDF1，UDF2。。。。UDF1xxx*/sparkSession.udf().register("StrLen",new UDF2<String, Integer, Integer>() {/****/private static final long serialVersionUID = 1L;@Overridepublic Integer call(String t1, Integer t2) throws Exception {return t1.length() + t2;}} ,DataTypes.IntegerType );sparkSession.sql("select name ,StrLen(name,100) as length from user").show(); // sparkSession.stop();

UDAF:用户自定义聚合函数。

• 实现UDAF函数如果要自定义类要实现UserDefinedAggregateFunction类

java:

SparkSession sparkSession = SparkSession.builder().appName("udaf").master("local").getOrCreate();JavaSparkContext sc = new JavaSparkContext(sparkSession.sparkContext()); JavaRDD<String> parallelize = sc.parallelize(Arrays.asList("zhansan","lisi","wangwu","zhangsan","zhangsan","lisi")); JavaRDD<Row> rowRDD = parallelize.map(new Function<String, Row>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Row call(String s) throws Exception {return RowFactory.create(s);} });List<StructField> fields = new ArrayList<StructField>(); fields.add(DataTypes.createStructField("name", DataTypes.StringType, true)); StructType schema = DataTypes.createStructType(fields); Dataset df = sparkSession.createDataFrame(rowRDD, schema); df.registerTempTable("user"); /*** 注册一个UDAF函数,实现统计相同值得个数* 注意：这里可以自定义一个类继承UserDefinedAggregateFunction类也是可以的*/ sparkSession.udf().register("StringCount", new UserDefinedAggregateFunction() {/*** */private static final long serialVersionUID = 1L;/*** 更新 可以认为一个一个地将组内的字段值传递进来 实现拼接的逻辑* buffer.getInt(0)获取的是上一次聚合后的值* 相当于map端的combiner，combiner就是对每一个map task的处理结果进行一次小聚合 * 大聚和发生在reduce端.* 这里即是:在进行聚合的时候，每当有新的值进来，对分组后的聚合如何进行计算*/@Overridepublic void update(MutableAggregationBuffer buffer, Row arg1) {buffer.update(0, buffer.getInt(0)+1);}/*** 合并 update操作，可能是针对一个分组内的部分数据，在某个节点上发生的 但是可能一个分组内的数据，会分布在多个节点上处理* 此时就要用merge操作，将各个节点上分布式拼接好的串，合并起来* buffer1.getInt(0) : 大聚和的时候 上一次聚合后的值 * buffer2.getInt(0) : 这次计算传入进来的update的结果* 这里即是：最后在分布式节点完成后需要进行全局级别的Merge操作*/@Overridepublic void merge(MutableAggregationBuffer buffer1, Row buffer2) {buffer1.update(0, buffer1.getInt(0) + buffer2.getInt(0));}/*** 指定输入字段的字段及类型*/@Overridepublic StructType inputSchema() {return DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("name", DataTypes.StringType, true)));}/*** 初始化一个内部的自己定义的值,在Aggregate之前每组数据的初始化结果*/@Overridepublic void initialize(MutableAggregationBuffer buffer) {buffer.update(0, 0);}/*** 最后返回一个和DataType的类型要一致的类型，返回UDAF最后的计算结果*/@Overridepublic Object evaluate(Row row) {return row.getInt(0);}@Overridepublic boolean deterministic() {//设置为truereturn true;}/*** 指定UDAF函数计算后返回的结果类型*/@Overridepublic DataType dataType() {return DataTypes.IntegerType;}/*** 在进行聚合操作的时候所要处理的数据的结果的类型*/@Overridepublic StructType bufferSchema() {return DataTypes.createStructType(Arrays.asList(DataTypes.createStructField("bf", DataTypes.IntegerType, true)));}});sparkSession.sql("select name ,StringCount(name) from user group by name").show();sparkSession.stop();

开窗函数

注意：

row_number() 开窗函数是按照某个字段分组，然后取另一字段的前几个的值，相当于 分组取topN

开窗函数格式：

row_number() over (partitin by XXX order by XXX)

java:

SparkSession sparkSession = SparkSession.builder().appName("window").master("local")//开启hive的支持，接下来就可以操作hive表了// 前提需要是需要开启hive metastore 服务.enableHiveSupport().getOrCreate();sparkSession.sql("use spark");sparkSession.sql("drop table if exists sales");sparkSession.sql("create table if not exists sales (riqi string,leibie string,jine Int) "+ "row format delimited fields terminated by 't'");sparkSession.sql("load data local inpath '/root/test/sales' into table sales");/*** 开窗函数格式：* 【 rou_number() over (partitin by XXX order by XXX) 】*/Dataset<Row> result = sparkSession.sql("select riqi,leibie,jine "+ "from ("+ "select riqi,leibie,jine,"+ "row_number() over (partition by leibie order by jine desc) rank "+ "from sales) t "+ "where t.rank<=3");result.show();sparkSession.stop();

总结

以上是生活随笔为你收集整理的spark-sql建表语句限制_SparkSQL的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。