【免费】离线数据处理任务一：数据抽取_大数据国赛sparkshell离线数据处理资源-CSDN文库

需积分: 0 172 浏览量 2024-01-15 10:53:58 上传评论收藏 109KB DOCX 举报

资源推荐

资源详情

资源评论

SPARK 增量抽取 MYSQL TO HIVE

陶运道 QQ：275931339

题目要求：

抽取 ds_db01 库中 customer_inf 的增量数据进入 Hive 的 ods 库中表 customer_inf。根据

ods.user_info 表中 modified_time 作为增量字段，只将新增的数据抽入，字段名称、类型

不变，同时添加静态分区，分区字段为 etl_date，类型为 String，且值为当前日期的前一

天日期（分区字段格式为 yyyyMMdd ）。使用 hive cli 执行 show partitions

ods.customer_inf 命令；

注：（1）mysql-connector-j-8.1.0.jar 包复制到 spark 的 jars 目录中。

（1）mysql 表中数据如图所示，按 modified_time 日期字段将表中数据全量或增量抽入到

hive 中。

全量抽入是将 19999 以前记录抽到 dwd 库中

增量抽入是将 19999 以后(包括 19999)记录抽取到 ods 库中，

select *

|from data where modified_time<to_timestamp('2023-1-1')

其中 data 是 MySQL 的临时表

（3） mysql 中有字段类型为 datatime imestamp 时导入到 hive 类型变化，所以要加以下

形式 jdbc:mysql://master:3306/ds_db01? serverTimezone=GMT

（4）大赛第 1 天全量抽取到 ods 库，并进行清洗，进入 dwd 库

第 2 天增量抽取并清洗进入 dwd 库

import org.apache.spark.sql.SparkSession

object scal_1 {

def main(args:Array[String])={

val spark = SparkSession.builder()

.master("spark://master:7077")

.appName("抽取数据")

.enableHiveSupport()

.config("spark.sql.warehouse.dir",

"hdfs://master:50070/usr/hive/warehouse")

.config("hive.metastore.uris", "thrift://master:9083")

.getOrCreate()

spark.read

.format("jdbc")

本内容试读结束，登录后可阅读更多

下载后可阅读完整内容，剩余6页未读，立即下载

内容反馈

taoyundao_1

粉丝: 126
资源: 2

最新资源

资源上传下载、课程学习等过程中有任何疑问或建议，欢迎提出宝贵意见哦~我们会及时处理！点击此处反馈

feedback-tip