编写Scala代码，使用Spark讲Mysql数据表中的数据抽取到Hive的ODS层

抽取MySQL的metast库中Production表的全量数据进入Hive的ods库中表production，字段排序、类型不变，同时添加静态分区，分区字段类型为String，且值为当前日期的前一天日期（分区字段格式为yyyyMMdd）。

使用IDEA创建maven项目

配置pom文件

<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>com.tledu</groupId><artifactId>llll</artifactId><version>1.0-SNAPSHOT</version><name>${project.artifactId}</name><description>My wonderfull scala app</description><inceptionYear>2018</inceptionYear><licenses><license><name>My License</name><url>http://....</url><distribution>repo</distribution></license></licenses><properties><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><encoding>UTF-8</encoding><scala.version>2.11.11</scala.version><scala.compat.version>2.11</scala.compat.version><spec2.version>4.2.0</spec2.version></properties><dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>${scala.version}</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_${scala.compat.version}</artifactId><version>2.3.2</version><scope>provided</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${scala.compat.version}</artifactId><version>2.3.2</version><scope>provided</scope></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.11</artifactId><version>2.0.2</version><scope>provided</scope></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.23</version></dependency><!-- Test --><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.12</version><scope>test</scope></dependency><dependency><groupId>org.scalatest</groupId><artifactId>scalatest_${scala.compat.version}</artifactId><version>3.0.5</version><scope>test</scope></dependency><dependency><groupId>org.specs2</groupId><artifactId>specs2-core_${scala.compat.version}</artifactId><version>${spec2.version}</version><scope>test</scope></dependency><dependency><groupId>org.specs2</groupId><artifactId>specs2-junit_${scala.compat.version}</artifactId><version>${spec2.version}</version><scope>test</scope></dependency></dependencies><build><sourceDirectory>src/main/scala</sourceDirectory><testSourceDirectory>src/test/scala</testSourceDirectory><plugins><plugin><!-- see http://davidb.github.com/scala-maven-plugin --><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.3.2</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals><configuration><args><arg>-dependencyfile</arg><arg>${project.build.directory}/.scala_dependencies</arg></args></configuration></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><version>2.21.0</version><configuration><!-- Tests will be run with scalatest-maven-plugin instead --><skipTests>true</skipTests></configuration></plugin><plugin><groupId>org.scalatest</groupId><artifactId>scalatest-maven-plugin</artifactId><version>2.0.0</version><configuration><reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory><junitxml>.</junitxml><filereports>TestSuiteReport.txt</filereports><!-- Comma separated list of JUnit test class names to execute --><jUnitClasses>samples.AppTest</jUnitClasses></configuration><executions><execution><id>test</id><goals><goal>test</goal></goals></execution></executions></plugin><plugin><artifactId>maven-assembly-plugin</artifactId><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>assembly</goal></goals></execution></executions></plugin></plugins></build>
</project>

导入scala
我这里演示用的是Unbanto，操作步骤一样

点击＋号去添加，这里注意scala版本号要与pom配置文件中的一致

创建一个scala目录并将它标记为根目录，在scala里新建一个object

编程过程如下

object demo01 {def getYesterday(): String = {val dateFormat: SimpleDateFormat = new SimpleDateFormat("yyyyMMdd")val cal: Calendar = Calendar.getInstance()cal.add(Calendar.DATE, -1)dateFormat.format(cal.getTime())}def main(args: Array[String]): Unit = {//source startval spark = SparkSession.builder().master("local[1]").config("spark.sql.parquet.writeLegacyFormat", true)//100个分区，执行完之后只有一个分区；.config("spark.sql.sources.partitionOverwriteMode", "dynamic")//动态分区.config("spark.sql.legacy.parquet.int96RebaseModeInWrite","LEGACY").config("hive.exec.dynamic.partition.mode", "nonstrict").enableHiveSupport().getOrCreate()//spark连接mysqlval url = s"jdbc:mysql://IP地址:3306/shtd_industry?useUnicode=true&characterEncoding=utf8&useSSL=false"val readerCustomerInf = spark.read.format("jdbc").option("url", url).option("driver", "com.mysql.jdbc.Driver").option("user", "root").option("password", "123456").option("dbtable", "数据库表名").load() //转换为DataFrame//source end//增加分区字段   etlval addPtDF = readerCustomerInf.withColumn("etl_date", lit(getYesterday()))val tableName = "hive表名"//切换hive的数据库import spark.sqlsql("use ods")//sinkaddPtDF.write.mode("overwrite").partitionBy("etl_date").saveAsTable(tableName).formatted("orc")spark.table(tableName).show()}
}

将编写好的代码打包发送到linux中

在集群上上传你打好的包

通常使用rz指令上传

可以写一个脚本运行你的包

vi spark.sh

/opt/module/spark-3.1.1-yarn/bin/spark-submit \
--class 要运行的类名 \
--master yarn \
--deploy-mode client \
--driver-memory 2g \
--executor-memory 1g \
--executor-cores 2 \
/jar包的地址/这里是你的jar包

保存退出

sh spark.sh 运行脚本
Mysql数据就导入HIVE数据库的ods层中了

编写Scala代码，使用Spark讲Mysql数据表中的数据抽取到Hive的ODS层相关推荐

MySql删除表中重复数据
有一表中存在大量重复数据在此记录下我删除表内重复数据的方法 -- 新增测试表 create table basic_farmer ( id INT(11), user_name VARCHAR(25 ...
python亿级mysql数据库导出_Python实现将MySQL数据库表中的数据导出生成csv格式文件的方法...
本文实例讲述了python实现将MySQL数据库表中的数据导出生成csv格式文件的方法.分享给大家供大家参考,具体如下: #!/usr/bin/env python # -*- coding:utf- ...
mysql中清空数据表中的数据，不删除数据表
1.清空不带外键约束的数据表中的数据使用delete语句清空`t_test`表中的数据 delete from `t_test`; 使用truncate语句清空`t_test`表中的数据 trunc ...
mysql删除表中所有数据的语句_sql删除数据库中所有表与数据语句
来源:转载如果要删除数据表中所有数据只要遍历一下数据库再删除就可以了,清除所有数据我们可以使用搜索出所有表名,构造为一条SQL语句进行清除了,这里我一一给各位同学介绍. 使用sql删除数据库中所有表 ...
MySQL删除表中的数据
Mysql删除表中的数据有三种方法,分别是delete ,drop,truncate. 一.delete删除表中的数据 delete好from结合使用,格式一般为:delete from 表名 whe ...
php怎么删除表数据,php怎样删除数据表中的数据_后端开发
php删除数据表中的数据的要领:能够经由过程mysqli_query()函数连系DELETE FROM语句来举行删除.DELETE FROM语句用于从数据库表中删除纪录,语法结构为:[DELETE F ...
17.2.3 通过查看triggers数据表中的数据查看触发器的信息
17.2.3 通过查看triggers数据表中的数据查看触发器的信息在MySQL中,会将触发器的信息存储到information_schema数据库中的triggers数据表中.可以通过查看info ...
数据库笔记03：管理数据表中的数据
/*************************** 第三单元:管理数据表中的数据 ***************************/ /************************* ...
xlsx表格怎么做汇总统计_Excel表格中如何快速汇总多个数据表中的数据
原标题:Excel表格中如何快速汇总多个数据表中的数据在Excel工作表中,如果需要汇总报告多个单独单元格的结果,可以将这些单元格中的数据合并到一个主工作表中.这些工作表可以与主工作表在同一个工作簿 ...

编写Scala代码，使用Spark讲Mysql数据表中的数据抽取到Hive的ODS层

编写Scala代码，使用Spark讲Mysql数据表中的数据抽取到Hive的ODS层

编写Scala代码，使用Spark讲Mysql数据表中的数据抽取到Hive的ODS层相关推荐

最新文章

热门文章