Sparksql与hive进行整合(代码中操作hive)

一环境准备

• 在代码中访问hive需要导入hive的依赖和jdbc依赖
• 需要hive的配置文件&＃xff0c;否则默认访问的是自带的hive
• spark-shell 中默认是开启支持hive的&＃xff0c;在本地代码中没有开启&＃xff0c;需手动开启

pom依赖

<dependencies><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-hive_2.11artifactId><version>2.1.1version>dependency>
<dependency><groupId>org.apache.sparkgroupId><artifactId>spark-core_2.11artifactId><version>2.1.1version>dependency><dependency><groupId>mysqlgroupId><artifactId>mysql-connector-javaartifactId><version>5.1.27version>dependency><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-sql_2.11artifactId><version>2.1.1version>dependency>
dependencies>


二 代码编写

object SparkSql2Hive {def main(args: Array[String]): Unit &＃61; {// 创建sparkSession 对象val spark: SparkSession &＃61; SparkSession.builder().appName("SparkSql2Hive").master("local[2]").enableHiveSupport() //开启支持hive.getOrCreate()spark.sql("show tables").show()spark.close()}
}


三 在此基础上基于二对表userinfo中的数据进行指标计算

spark.sql("use default")// 都是懒执行的只有在调用行动算子后才会执行spark.sql("select * from userinfo").show()


数据如下&＃xff1a;

&＃43;---&＃43;---------&＃43;----&＃43;
|u01|2017/1/21| 5|
|u02|2017/1/23| 6|
|u03|2017/1/22| 8|
|u04|2017/1/20| 3|
|u01|2017/1/23| 6|
|u01|2017/2/21| 8|
|U02|2017/1/23| 6|
|U01|2017/2/22| 4|
| | null|null|
&＃43;---&＃43;---------&＃43;----&＃43;


需求&＃xff1a;
① 过滤掉空的数据
② 计算出每个用户累计到当前月的充值金额
思路&＃xff1a;开窗累加&＃xff0c;按照用户id进行分区&＃xff0c;按照时间排序&＃xff0c;指定窗口大小为上无边界&＃xff0c;到当前行
③ 实现&＃xff1a;

spark.sql("""|select|lower(uid) as uid,|regexp_replace(dt,&＃39;(\\/&＃43;)&＃39;,"-") as dayTime,|num|from userinfo|where regexp_replace(dt,&＃39;(\\/&＃43;)&＃39;,"-") is not null""".stripMargin).createOrReplaceTempView("t1")spark.sql("""|select|uid,|dayTime,|num,|sum(num)over(partition by uid order by dayTime asc rows between unbounded preceding and current row)|from t1""".stripMargin).show()


指标分析结果&＃xff1a;

&＃43;---&＃43;---------&＃43;---&＃43;------------------------------------------------------------------------------------------------------------------&＃43;
|uid| dayTime|num|sum(num) OVER (PARTITION BY uid ORDER BY dayTime ASC NULLS FIRST ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)|
&＃43;---&＃43;---------&＃43;---&＃43;------------------------------------------------------------------------------------------------------------------&＃43;
|u01|2017-1-21| 5| 5|
|u01|2017-1-23| 6| 11|
|u01|2017-2-21| 8| 19|
|u01|2017-2-22| 4| 23|
|u02|2017-1-23| 6| 6|
|u02|2017-1-23| 6| 12|
|u04|2017-1-20| 3| 3|
|u03|2017-1-22| 8| 8|
&＃43;---&＃43;---------&＃43;---&＃43;------------------------------------------------------------------------------------------------------------------&＃43;


三 总结

在创建Spark-Session对象的时候使用 .enableHiveSupport() //开启支持hive
注意&＃xff1a;
在使用此种方式创建hive的数据库的时候会出现问题&＃xff0c;需要增加配置
.config(“spark.sql.warehouse.dir”, “hdfs://hadoop102:9000/user/hive/warehouse”)
创建SparkSession对象的语句如下&＃xff1a;

val spark: SparkSession &＃61; SparkSession.builder().appName("SparkSql2Hive").master("local[2]").enableHiveSupport() //开启支持hive.config("spark.sql.warehouse.dir", "hdfs://hadoop102:9000/user/hive/warehouse").getOrCreate()