Spark操作Hive误区
最近在筹备大数据竞赛，在训练的时候出现了挺多问题，现在统一总结一下
出现的问题
首先，新版IDEA不能从正常的创建项目哪里创建Maven项目（官方的打包方法为sbt,所以正在逐渐淘汰Maven）
但是训练时候依旧需要使用Maven去构建项目，所以经查资料创建方法如下
首先创建一个Scala项目，构建系统选择Intellij，选择创建。
右击该项目，选择添加框架支持
勾选Maven，点击确定
此时已经创建好pom.xml文件了
这时就可以添加对应的依赖了
拉取对应的依赖
依赖文件如下
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>top.smallway</groupId>
<artifactId>sparkapp</artifactId>
<version>1.0</version>
<properties>
<scala.version>2.12.16</scala.version>
</properties>
<dependencies>
<!--scala-->
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.version}</version>
</dependency>
<!--spark-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.32</version>
</dependency>
<!-- spark-hive-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<!-- hive-->
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-jdbc</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>com.github.housepower</groupId>
<artifactId>clickhouse-integration-spark_2.12</artifactId>
<version>2.5.4</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-auth</artifactId>
<version>3.2.1</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.10.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.10.0</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<testSourceDirectory>src/test/scala</testSourceDirectory>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
</project>此时重新拉取Maven即可
编写Scala代码EltUtil.scala类
import org.apache.spark.sql._
import org.apache.spark.sql.jdbc.{ClickHouseDialect, JdbcDialects}
// ELT工具类
object EltUtil {
// extract from mysql
/**
* @param spark SparkSession实例
* @param jdbcMap 要加载的JDBC配置项
*
* @return 返回一个DataFrame（Dataset[Row]）
*
*/
def extractFromJDBC(spark: SparkSession, jdbcMap:Map[String,String]): Dataset[Row] = {
// 读取JDBC数据源，创建DataFrame
val df = spark
.read
.format("jdbc")
.options(jdbcMap)
.load()
// 返回
df
}
// load to hive
/**
* @param spark
* @param df 要装载到Hive中的DataFrame
* @param db 要装载到的Hive数据库
* @param tb 要装载到的Hive ODS表
* @param partitionColumn 指定分区列
*
* @return unit
*/
def loadToHive(spark: SparkSession, df:Dataset[Row], hiveMap: Map[String,String]):Unit = {
val database = hiveMap("db") // 要写入的数据库
val table = hiveMap("tb") // 要写入的数据表
val partitionColumn = hiveMap.get("partitionColumn") // 分区列
spark.sql(s"use $database") // 打开指定数据库，这里使用了字符串插值
// 有的表需要分区，有的不需要。这里使用模式匹配来分别处理
partitionColumn match{
case Some(column) =>
df.write
.format("parquet")
.mode("overwrite") // 覆盖
.partitionBy(column) // 指定分区
.saveAsTable(table)
case None =>
df.write
.format("parquet")
.mode("overwrite") // 覆盖
.saveAsTable(table) // saveAsTable()方法：会将DataFrame数据保存到Hive表中
}
}
// 增量写入方法
def appendToHive(spark: SparkSession, df:Dataset[Row], hiveMap: Map[String,String]):Unit = {
val database = hiveMap("db") // 要写入的数据库
val table = hiveMap("tb") // 要写入的数据表
val partitionColumn = hiveMap.get("partitionColumn") // 分区列
spark.sql(s"use $database") // 打开指定数据库，这里使用了字符串插值
// 有的表需要分区，有的不需要。这里使用模式匹配来分别处理
partitionColumn match{
case Some(column) =>
df.write
.format("parquet")
.mode("append") // 追加
.partitionBy(column) // 指定分区
.saveAsTable(table)
case None =>
df.write
.format("parquet")
.mode("append") // 追加
.saveAsTable(table) // saveAsTable()方法：会将DataFrame数据保存到Hive表中
}
}
// 定义一个函数，用来将分析结果导出到mysql中
/**
*
* @param db 目标数据表
* @param df 分析结果集
*/
import org.apache.spark.sql._
import java.util.Properties
def exportToMysql(df:Dataset[Row], tb:String) = {
val DB_URL= "jdbc:mysql://localhost:3306/shtd_result?useSSL=false" // 数据库连接url，请将localhost替换为服务器所在IP
// 下面创建一个prop 变量用来保存JDBC 连接参数
val props = new Properties()
props.put("user", "root") // 表示用户名是root
props.put("password", "admin") // 表示密码是hadoop
props.put("driver","com.mysql.jdbc.Driver") // 表示驱动程序
df.write.mode("overwrite").jdbc(DB_URL, tb, props)
}
// 定义一个函数，用来将分析结果导出到clickhouse中
/**
*
* @param db 目标数据表（形式：db.tb）
* @param df 分析结果集
*/
def exportToClickHouse(df:Dataset[Row], table:String) = {
// 注册ClickHouseDialect
JdbcDialects.registerDialect(ClickHouseDialect)
// clickhouse驱动程序和连接信息
val ckDriver = "com.github.housepower.jdbc.ClickHouseDriver" // 驱动程序
val ckUrl = "jdbc:clickhouse://localhost:9000" // 数据库连接url，请将localhost替换为服务器所在IP
val ckUser = "default"
val ckPassword = ""
// 写出
try {
val props = new java.util.Properties
props.put("driver", ckDriver)
props.put("user", ckUser)
props.put("password", ckPassword)
df.write
.mode("append")
.option("batchsize", "10000")
.option("isolationLevel", "NONE")
.option("numPartitions", "1")
.jdbc(ckUrl, table, props)
} catch {
case e:Exception =>
println("catch and ignore!")
println(e)
}
}
}在主方法中调用即可
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.lit
object spark {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder()
.master("local[*]")
.appName("sparkapp")
.enableHiveSupport()
.getOrCreate()
val stock_customer_info_sql = "select * from customer_inf where modified_time <= '2022-08-20 23:59:59'"
val jdbcMap = Map(
"url" -> "jdbc:mysql://localhost/database",
"user" -> "user",
"password" -> "password",
"query" -> stock_customer_info_sql,
)
val dataDF = EltUtil.extractFromJDBC(spark, jdbcMap)
val df12 = dataDF.withColumn("etl_date", lit("20221123"))
val hiveMap = Map(
"db" -> "Hive数据库",
"tb" -> "Hive数据库中的表",
"partitionColumn" -> "列名"
)
EltUtil.loadToHive(spark, df12, hiveMap)
spark.table("Hive数据库.表名").show()
}
}