SparkSQL入门

入门"/>

SparkSQL入门

概述

两种模式

Spark on Hive: 语法是Spark SQL语法，实际上是在IDEA上编写java叠加SQL的代码。

Hive on Spark: 只是替换了Hadoop的MR，改为了Spark的计算引擎。

发展历史

RDD => DataFrame => DataSet：

1. 都有惰性机制，遇到行动算子才会执行。
2. 三者都会根据Spark的内存情况自动缓存运算
3. 三者都有分区的概念

特点

1. 易整合：无缝的整合了SQL查询和Spark编程
2. 统一的数据访问方式：使用相同的方式连接不同的数据源
3. 兼容Hive：在已有的仓库上直接运行SQL或者HQL
4. 标准的数据连接：通过JDBC或者ODBC来连接

数据的加载和保存

json文件：spark数据读取时，读取后会自动解析JSON，并且附加上列名和属性类型。并且兼容RDD的算子操作，

public class SQL_Test {public static void main(String[] args) {SparkConf sparkConf = new SparkConf().setAppName("SparkSQL").setMaster("local[*]");SparkSession spark = SparkSession.builder().config(sparkConf).getOrCreate();DataFrameReader read = spark.read();//读取后会自动解析JSON，并且附加上列名和属性类型Dataset<Row> userJSON = read.json("input/user.json");//打印数据类型userJSON.printSchema();userJSON.show();//即收集又打印spark.close();}
}

自定义函数

UDF操作单个数据，产生单个数据

import org.apache.spark.SparkConf;
import org.apache.spark.sql.*;
import org.apache.spark.sql.api.java.UDF2;
import org.apache.spark.sql.types.DataTypes;

/**

• title:

• @Author 浪拍岸

• @Create 19/10/2023 上午8:52

• @Version 1.0
  /
  public class SQL_UDF {
  public static void main(String[] args) {
  SparkConf sparkConf = new SparkConf().setAppName(“SparkSQL”).setMaster("local[]");
  SparkSession spark = SparkSession.builder().config(sparkConf).getOrCreate();

   DataFrameReader read = spark.read();//读取后会自动解析JSON，并且附加上列名和属性类型Dataset<Row> userJSON = read.json("input/user.json");userJSON.createOrReplaceTempView("t1");//注册函数spark.udf().register("myudf", new UDF2<String, Long, String>() {@Overridepublic String call(String s, Long integer) throws Exception {if(integer >= 18){return s+"大侠";}else{return s+"小虾米";}//return null;}}, DataTypes.StringType);//spark.sql("select myudf(name,age) from t1").show();spark.close();
  

  }
  }

UDAF操作多个数据，产生单个数据

import org.apache.spark.SparkConf;
import org.apache.spark.sql.*;
import org.apache.spark.sql.api.java.UDF2;
import org.apache.spark.sql.types.DataTypes;
import static org.apache.spark.sql.functions.udaf;public class SQL_UDAF {public static void main(String[] args) {SparkConf sparkConf = new SparkConf().setAppName("SparkSQL").setMaster("local[*]");SparkSession spark = SparkSession.builder().config(sparkConf).getOrCreate();DataFrameReader read = spark.read();//读取后会自动解析JSON，并且附加上列名和属性类型Dataset<Row> userJSON = read.json("input/user.json");userJSON.createOrReplaceTempView("t1");//注册函数spark.udf().register("ageAVG", udaf(new AgeAvg(), Encoders.LONG()));//spark.sql("select name,ageAVG(age) from t1 group by name").show();spark.close();}
}

文件的读取和转换保存

1. json格式
2. csv格式
3. parquet、orc格式

Hive交互

1. 开启Hive支持enableHiveSupport()
2. 用户权限造假System.setProperty("HADOOP_USER_NAME","atguigu");
3. 添加hive-site.xml到resource目录下

import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;/*** title:** @Author 浪拍岸* @Create 19/10/2023 下午3:35* @Version 1.0*/
public class HiveTest {public static void main(String[] args) {System.setProperty("HADOOP_USER_NAME","atguigu");SparkConf sparkConf = new SparkConf().setAppName("SparkSQL").setMaster("local[*]");SparkSession spark = SparkSession.builder().enableHiveSupport().config(sparkConf).getOrCreate();//        spark.sql("show tables").show();spark.sql("select * from stu where id = 1").createOrReplaceTempView("t1");spark.sql("select * from t1").show();spark.close();}
}