编写 Scala 工程代码，将 MySQL 库中的表增量抽取到 Hive库中对应表中-Toy模板网

这篇具有很好参考价值的文章主要介绍了编写 Scala 工程代码，将 MySQL 库中的表增量抽取到 Hive库中对应表中。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

提示：本文采用IDEA编写代码

操作步骤：

一、在linux上

搭建好spark，推荐一篇较好的文章：spark3.3.0安装&部署过程。

注：如果需要运行yarn模式，在spark-env.sh文件末尾添加：
export YARN_CONF_DIR=/opt/hadoop-3.1.3/etc/hadoop
export HADOOP_CONF_DIR=/opt/hadoop-3.1.3/etc/hadoop

搭建完spark集群后，在spark的jars目录下放入mysql驱动jar包。
在spark的conf目录下放入hive-site.xml文件。

二、在IDEA上

1.创建一个maven项目

编写 Scala 工程代码，将 MySQL 库中的表增量抽取到 Hive库中对应表中

2.安装scala插件

编写 Scala 工程代码，将 MySQL 库中的表增量抽取到 Hive库中对应表中

3.在项目结构的全局库导入scala至项目

编写 Scala 工程代码，将 MySQL 库中的表增量抽取到 Hive库中对应表中

4.在pom文件导入以下两个依赖

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.3.1</version>
        </dependency>
    </dependencies>

5.编写代码

涉及到hive3.x版本的动态分区，仅供参考

package com.test

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import java.text.SimpleDateFormat
import java.util.{Calendar, Properties}

object lx1 {
  def main(args: Array[String]): Unit = {
    //    建立和spark框架的连接及定义业务名称
    val conf = new SparkConf().setMaster("local[*]").setAppName("数据抽取")
    val spark = SparkSession.builder().enableHiveSupport().config(conf).getOrCreate()
    //spark读取mysql
    val df = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://master:3306/test?useUnicode=true&characterEncoding=utf-8")
      .option("driver", "com.mysql.cj.jdbc.Driver")
      .option("user", "root")
      .option("password", "123")
      .option("dbtable", "orders")
      .load()
    df.createOrReplaceTempView("v_orders")
//    获取时间
    val day = Calendar.getInstance()
    day.add(Calendar.DATE, -2)
    val sdf = new SimpleDateFormat("yyyy-MM-dd")
    val yesDate = sdf.format(day.getTime)
    //开启动态分区功能
    spark.sqlContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
    //把读取的数据写入hive
    spark.sql("insert into table test.d_p_t2 " +
      "partition(custkey) select orderkey, orderdate,custkey " +
      " from v_orders where orderdate='" + yesDate + "'")
    spark.stop()
  }
}