【黑马甄选离线数仓day02

1. 数仓工具使用-DataX

1.1 DataX介绍

DataX 是阿里推出的一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。

将DataX安装好之后, 仅需要配置Json的采集文件即可实现数据的同步

在这里插入图片描述

DataX几个组成部分:

Reader：Reader为数据采集模块，负责采集数据源的数据，将数据发送给Framework。
Writer：Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
Framework：Framework用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。

在这里插入图片描述

经过几年积累，DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、NOSQL、大数据计算系统都已经接入。DataX目前支持数据如下：

在这里插入图片描述

DataX的国内的开源的地址: https://gitee.com/mirrors/DataX?_from=gitee_search

DataX的GitHub的开源地址: https://github.com/alibaba/DataX

1.2 DataX-命令使用

1.2.1 MySQL写入HDFS

准备mysql测试数据

1. 在mysql创建test数据库，在此数据库下创建student表，然后往表中插入数据。

-- 建库
create database if not exists test character set utf8;
-- 使用库
use test;
-- 建表
create table student(id int,name varchar(20),age int,createtime timestamp );
-- 插入数据
insert into `student` (`id`, `name`, `age`, `createtime`) values('1','zhangsan','18','2021-05-10 18:10:00');
insert into `student` (`id`, `name`, `age`, `createtime`) values('2','lisi','28','2021-05-10 19:10:00');
insert into `student` (`id`, `name`, `age`, `createtime`) values('3','wangwu','38','2021-05-10 20:10:00');
-- 验证数据
select id,name,age,createtime from student where age <30;

在这里插入图片描述

确定hdfs目标位置

创建目标路径: /test/datax/mysql2hdfs (可选的,因为会自动生成)

[root@hadoop01 job]# hdfs dfs -mkdir -p /test/datax/mysql2hdfs

在这里插入图片描述

创建datax_json文件

方式1: 命令

命令查看配置模板

方式1: 执行以下脚本命令，查看基础配置模板。

[root@hadoop01 ~]# cd /export/server/datax

[root@hadoop01 datax]# python bin/datax.py -r mysqlreader -w hdfswriter

创建配置文件

[root@hadoop01 datax]# cd /export/server/datax/job

[root@hadoop01 job]# vim mysql2hdfs.json

添加以下内容到mysql2hdfs.json

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "column": ['id','name','age','createtime'],
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:mysql://hadoop01:3306/test"],
                                "table": ["student"]
                            }
                        ],
                        "password": "123456",
                        "username": "root",
                        "where": "age <30"
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "column": [
							{'name':'id','type':'int'}
							{'name':'name','type':'string'}
							{'name':'age','type':'int'}
							{'name':'createtime','type':'timestamp'}

						],

                        "defaultFS": "hdfs://hadoop01:8020",
                        "fieldDelimiter": ",",
                        "fileName": "student",
                        "fileType": "text",
                        "path": "/test/datax/mysql2hdfs",
                        "writeMode": "append"
                    }
                }
            }
        ],
        "setting": {
            "speed": {
                "channel": "1"
            }
        }
    }
}

方式2: 官方

方式2: 浏览器打开以下链接,查看所有配置模板

链接: https://gitee.com/mirrors/DataX/tree/master

创建配置文件

[root@hadoop01 datax]# cd /export/server/datax/job

[root@hadoop01 job]# vim mysql2hdfs.json

添加如下内容到mysql2hdfs.json文件中:

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "querySql": [
                                    "select id,name,age,createtime from student where age <30;"
                                ],
                                "jdbcUrl": [
                                    "jdbc:mysql://hadoop01:3306/test"
                                ]
                            }
                        ]
                    }
                }, 
                "writer":{
                    "name": "hdfswriter",
                    "parameter": {
                        "defaultFS": "hdfs://hadoop01:8020",
                        "fileType": "text",
                        "path": "/test/datax/mysql2hdfs",
                        "fileName": "student.txt",
                        "column": [
                            {
                                "name": "id",
                                "type": "INT"
                            },
                            {
                                "name": "name",
                                "type": "STRING"
                            },
                            {
                                "name": "age",
                                "type": "int"
                            },
                            {
                                "name": "createtime",
                                "type": "TIMESTAMP"
                            }
                        ],
                        "writeMode": "append",
                        "fieldDelimiter": "\t",
                        "compress":"GZIP"
                    }
                }
            }
        ], 
        "setting": {
            "speed": {
                "channel": "1"
            }
        }
    }
}

启动并运行dataX

[root@hadoop01 job]# cd /export/server/datax/
[root@hadoop01 datax]# python bin/datax.py job/mysql2hdfs.json

| 在这里插入图片描述

校验是否成功

如果是压缩的建议下载,解压查看数据是否成功导入hdfs

在这里插入图片描述

解压, 查看内容即可:

在这里插入图片描述

练习

准备练习数据:

-- 建表
create table if not exists product(
    id int,
    name varchar(20),
    brand varchar(20),
    color varchar(20)
);
-- 插入数据
insert into product (id, name, brand, color)
values
       (1,'iphone 15','iphone','yellow'),
       (2,'huawei meta 60','huawei','white'),
       (3,'xiaomi 13','xiaomi','yellow');
-- 验证数据
select id,name,brand,color from product where color = 'yellow';

创建json文件

[root@hadoop01 datax]# cd job

[root@hadoop01 job]# vim mysql2hdfs_lx.json

添加以下内容:

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "querySql": [
                                    "select id,name,brand,color from product where color = 'yellow';"
                                ],
                                "jdbcUrl": [
                                    "jdbc:mysql://hadoop01:3306/test"
                                ]
                            }
                        ]
                    }
                }, 
                "writer":{
                    "name": "hdfswriter",
                    "parameter": {
                        "defaultFS": "hdfs://hadoop01:8020",
                        "fileType": "text",
                        "path": "/test/datax/mysql2hdfs_lx",
                        "fileName": "product.txt",
                        "column": [
                            {
                                "name": "id",
                                "type": "INT"
                            },
                            {
                                "name": "name",
                                "type": "STRING"
                            },
                            {
                                "name": "brand",
                                "type": "STRING"
                            },
                            {
                                "name": "color",
                                "type": "STRING"
                            }
                        ],
                        "writeMode": "append",
                        "fieldDelimiter": "\t"
                    }
                }
            }
        ], 
        "setting": {
            "speed": {
                "channel": "1"
            }
        }
    }
}

确定HDFS路径

/test/datax/mysql2hdfs_lx

启动datax

[root@hadoop01 job]# cd /export/server/datax/
[root@hadoop01 datax]# python bin/datax.py job/mysql2hdfs_lx.json

校验是否成功

1.2.2 MySQL写入HIVE

准备mysql测试数据

由于上述已经在mysql创建test数据库，在此数据库下创建了student表并往表中插入了数据,本次直接使用即可

在这里插入图片描述

确定hive目标位置

注意：Hive上设置的数据仓库的存储路径为：/user/hive/warehouse/库名.db/表名

建立测试数据库：test，

创建测试表：student；

则对应数据存储目标位置为：/user/hive/warehouse/test.db/student

-- 建立测试数据库：test
create database if not exists test;
-- 使用库
use test;
-- 创建测试表：student
create table student(
    id int comment '编号',
    name string comment '名称',
    age int comment '年龄',
    createtime string comment '创建时间'
)comment '测试表'
    row format delimited
fields terminated by ',';
-- 验证数据
select * from student limit 10;
-- 注意:对应表的存储路径为：/user/hive/warehouse/test.db/student

创建datax_json文件

创建作业的配置文件

进入/export/server/datax/job目录, 创建配置文件mysql2hive.json

[root@hadoop01 datax]# cd /export/server/datax/job

[root@hadoop01 job]# vim mysql2hive.json

注意：写入hive时与写入hdfs的写入名称name都为hdfswriter。只是在写入hive时，为与hive表关联，在path参数后填写hive表在hdfs上的存储路径。以及保证存储格式和分隔符号与HIVE表保持一致!!!

配置内容为:

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "querySql": [
                                    "select id,name,age,createtime from student where age <30;"
                                ],
                                "jdbcUrl": [
                                    "jdbc:mysql://hadoop01:3306/test"
                                ]
                            }
                        ]
                    }
                }, 
                "writer":{
                    "name": "hdfswriter",
                    "parameter": {
                        "defaultFS": "hdfs://hadoop01:8020",
                        "fileType": "text",
                        "path": "/user/hive/warehouse/test.db/student",
                        "fileName": "student.txt",
                        "column": [
                            {
                                "name": "id",
                                "type": "INT"
                            },
                            {
                                "name": "name",
                                "type": "STRING"
                            },
                            {
                                "name": "age",
                                "type": "int"
                            },
                            {
                                "name": "createtime",
                                "type": "TIMESTAMP"
                            }
                        ],
                        "writeMode": "append",
                        "fieldDelimiter": ",",
                        "compress":"GZIP"
                    }
                }
            }
        ], 
        "setting": {
            "speed": {
                "channel": "1"
            }
        }
    }
}

启动并运行dataX

[root@hadoop01 job]# cd /export/server/datax/
[root@hadoop01 datax]# python bin/datax.py job/mysql2hive.json

在这里插入图片描述

校验是否成功

如果是压缩的建议下载,解压查看数据是否成功导入hdfs

在这里插入图片描述

1.3 DataX-Web工具

DataX Web是在DataX之上开发的分布式数据同步工具，提供简单易用的操作界面，降低用户使用DataX的学习成本，缩短任务配置时间，避免配置过程中出错。

1.3.1 DataX-Web启动:

[root@hadoop01 datax]# cd /export/server/datax-web-2.1.2

[root@hadoop01 datax-web-2.1.2]# ./bin/start-all.sh

在这里插入图片描述

1.3.2 DataX-Web登录:

登录主界面: http://hadoop01:9527/index.html

用户名: admin 
密码: 123456

在这里插入图片描述

1.3.3 配置操作

执行器配置: 点击执行器管理。

执行器列表中显示在线的执行器列表, 可通过"在线机器"查看对应执行器的集群机器

在这里插入图片描述

创建数据源: 创建数据源（源库和目标库）。首先点击数据源管理，点击添加

在这里插入图片描述

再次点击添加，添加目标数据库信息。如下：

在这里插入图片描述

最终效果

在这里插入图片描述

创建项目: 点击项目管理，点击添加。创建测试项目

在这里插入图片描述

创建任务模板: 点击任务管理，点击DataX任务模板，点击添加

在这里插入图片描述

1.3.4 写入到Hive中

hive建表
- 在hive创建student2表

-- 使用库
use test;
-- 创建测试表：student
create table student2(
    id int comment '编号',
    name string comment '名称',
    age int comment '年龄',
    createtime string comment '创建时间'
)comment '测试表'
    row format delimited
fields terminated by ',';
-- 验证数据
select * from student2 limit 20;
-- 注意:对应表的存储路径为：/user/hive/warehouse/test.db/student2

构建json脚本

在这里插入图片描述

点击完成即可

点击任务管理, 执行任务

在这里插入图片描述

等待一会,任务执行成功, 若执行失败, 可以在点击下拉按钮后, 点击查询日志

在这里插入图片描述

如果有问题, 可以点击编辑修改配置

在这里插入图片描述

查看Hive表, 是否成功写入

2. 基于DataX完成数据导入

2.1 建模设计: 需要考虑的问题

在这里插入图片描述

1- 数据的同步方式是什么? 
	全量覆盖同步: 
		在建表的时候, 不需要构建分区表, 每一次都是将之前的数据全部删除, 然后全部都重新导入一遍
		适合于: 数据量比较少, 而且不需要维护历史变化行为
	
	仅新增同步:
		在建表的时候, 需要构建分区表, 分区字段是以更新的周期一致即可, 比如 更新的周期为天, 分区字段也应该为天, 每一次导入上一天的新增的数据
		适合于: 数据量比较大, 而且不需要维护历史变化行为(并不代表表不存在变化, 只不过这个变化对分析没有影响)
	
	新增及更新同步:
		处理逻辑: 在建表的时候, 需要构建分区表, 分区字段是以更新的周期一致, 比如 更新的周期为天, 分区字段也应该为天,每一次导入上一天的新增及更新的数据
		适合于: 数据量比较大, 而且需要后期维护历史变化
	
	全量同步:
		在建表的时候, 需要构建分区表, 分区字段以更新的周期一致即可, 比如 更新的周期为天, 分区字段也应该为天,每一次导入的时候, 都是将整个数据集全部导入到一个新的分区中, 后期定期删除老的历史数据(比如: 仅保留最近一周)
		适合于: 数据量比较少, 而且还需要维护历史变化, 同时维度周期不需要特别长
		注意: 此种同步方式相对较少


2- 表是否选择为内部表 还是 外部表?  
	判断的依据: 是否对数据有绝对的控制权, 如果没有 必须是外部表, 如果有 随意
	

3- 表是否为分区表还是分桶表?
	分区表: 分文件夹, 将数据划分到不同的文件夹中, 当查询数据的时候, 通过分区字段获取对应分区下的数据, 从而减少数据扫描量, 提高查询效率
	
	分桶表: 分文件 将数据根据指定的字段划分为N多个文件 可以通过这种方式对数据进行采样操作 以及分桶表在后续进行join优化的时候也会涉及到(bucket Map Join | SMB Join)

4- 表选择什么存储格式 和 压缩方案?
	存储格式: 一般都是 ORC / Text File
	压缩格式: 一般都会 SNAPPY / GZ
	
	存储格式:  如果数据直接对接的普通文本文件的操作 只能使用textFile 否则大多数都是ORC
	压缩格式:  读多写少 采用SNAPPY   写多读少 采用GZ  如果普通的文本文件对接, 一般不设置压缩
	
	如果空间比较充足, 没有特殊要求, 建议统一采用SNAPPY

5- 表中字段应该如何选择呢?  
	ODS层: 业务库有那些表, 表中有哪些字段, 对应在ODS层建那些表, 表中对应有相关的字段 额外根据同步方式, 选择是否添加分区字段
	其他层次: 不同层次 需要单独分析, 目标: 把需求分析的结果能够完整的在表中存储起来即可

缓慢渐变维:

SCD1:不维护历史变更行为, 直接对过去数据进行覆盖即可 
此种操作 仅适用于错误数据的处理

SCD2: 维护历史变化行为, 每天同步全量的数据, 不管是否有变化, 均全量维护到一个新的分区中

SCD3:维护历史变更行为,处理方式在表中新增两个新的字段,一个是起始时间,一个是结束时间,当数据发生变更后,将之前的数据设置为过期,将新的变更后完整的数据添加到表中,重新记录其起始和结束时间,将这种方案称为拉链表 
好处:可以维护更多的历史版本的数据, 处理起来也是比较简单的 (利于维护)
弊端:造成数据冗余存储 大量占用磁盘空间


SCD4:维度历史变化,处理方式,当表中有字段发生变更后,新增一列,将变更后的数据存储到这一列中即可 
好处:减少数据冗余存储
弊端:只能维护少量的历史版本, 而且维护不方便, 效率比较低

2.2 Hive中构建原始业务表

表设计:

表类型: 内部表 + ORC + ZLIB + 基于同步方式选择分区表
字段内容: 与业务库保持一致(注意调整数据类型)

在这里插入图片描述

参考建表脚本: <<黑马甄选离线数仓项目_完整资料\07_脚本\01_原始数据层建表SQL脚本>>

2.3 基于DataX完成数据导入

由于ODS层涉及到不同的同步方式导入方案, 每一种导入方式均有所差异性, 此处举例四种导出模式, 其余的表自行完成即可

初始准备工作

1- 在DataX-Web中新建项目: 黑马甄选大数据平台

在这里插入图片描述

2- 创建任务模板: 在任务管理的Datax任务模板中添加任务模板(每日凌晨执行一次)

在这里插入图片描述

添加完成，点击下次触发时间下的查看，可以看到后续执行的时间。

2.3.1 全量覆盖方案

1- 添加数据源: 点击数据源管理，点击添加。

首先添加mysql的dim库。

在这里插入图片描述

接着添加hive的dim库。

在这里插入图片描述

4- 任务构建。

在这里插入图片描述

构建writer时，writeMode应该为truncate，即如果目录下有fileName前缀的文件，先删除后写入。但是这里选项中没有，可以先选append，在之后构建json的时候进行更改。

在这里插入图片描述

更改writeMode为truncate

在这里插入图片描述

5- 执行任务

点击任务管理，可以看到刚刚构建的任务。如果没有，可以刷新一下网页。

点击状态下的启动，到达下次触发时间后，会自动执行任务。

也可以点击操作下的执行一次，立即执行任务。
之后查看日志，可以看到执行成功。
6- 查看Hive数据

查看HDFS

在这里插入图片描述

查看HIVE

在这里插入图片描述

经过核对，与mysql数据一致。
此时再次执行任务，执行成功后，发现hive表数据量没有变，说明同步方式是全量覆盖，即插入数据时先清空，后插入。

在这里插入图片描述

2.3.2 仅新增方案

仅新增方式: 订单退款表为例, 探讨如何完成仅新增方式导入操作

从业务库将数据导入到ODS层, 分为首次导入和增量导入两部分, 其中首次导入指的第一次建表, 导入数据, 此时一般都是全量导入, 后续每一天都是采用增量导入的方式,

当前项目, 增量模式: T+1(当天处理都是上一天的数据/ 每天的数据在下一天进行处理)

技术: DataX

思考: 当全量导入数据的时候, 整个数据集是应该放置到上一天的分区中呢? 还是说应该按照实际表数据创建时间划分到不同分区呢?

二种方式均可以, 其实在实际应用中两种其实都是存在的, 只不过我们当前选择的第二种方式, 直接将数据不同的日期放置到不同的分区下


如何做呢? DataX不支持直接导入多个分区数据, 仅支持导入一个分区, 因为DataX本质上是将数据对接HDFS, 而非HIVE,只是HIVE正好映射到HDFS对应目录下, 正好吧数据加载到了, 所以在HIVE正好就看到了数据
	
既然dataX不支持那如何处理呢? 
	先创建一个临时表(没有分区), 通过DataX 将数据导入到临时表中, 然后在通过临时表灌入到HIVE的ODS层目标表表, 处理完成后, 删除临时表

首次导入:

1- 创建订单退款表的临时表:

-- 构建订单退款表 临时表:
drop table if exists ods.ods_sale_shop_refund_i_temp;
CREATE  TABLE IF NOT EXISTS ods.ods_sale_shop_refund_i_temp(
    id                    BIGINT COMMENT '主键',
    refund_no             STRING COMMENT '退款单号',
    refund_status         BIGINT COMMENT '退款状态：1-退款中；2-退款成功；3-退款失败',
    refund_code           BIGINT COMMENT '退款原因code',
    refund_msg            STRING COMMENT '退款原因',
    refund_desc           STRING COMMENT '退款描述',
    create_time           TIMESTAMP COMMENT '创建时间/退款申请时间',
    update_time           TIMESTAMP COMMENT '更新时间',
    cancel_time           TIMESTAMP COMMENT '退款申请取消时间',
    refund_amount         DECIMAL(27, 2) COMMENT '退款金额',
    refund_point_amount   DECIMAL(27, 2) COMMENT '扣减已赠积分',
    return_pay_point      BIGINT COMMENT '退还支付积分',
    return_point_amount   DECIMAL(27, 2) COMMENT '退还积分抵扣金额',
    refund_time           TIMESTAMP COMMENT '退款成功时间',
    less_weight           DECIMAL(27, 3) COMMENT '差额重量,单位kg',
    pick_weight           DECIMAL(27, 3) COMMENT '拣货重量,单位kg',
    is_deleted            BIGINT COMMENT '失效标志：0-正常；1-失效',
    refund_type           BIGINT COMMENT '退款类型：1-部分退；2-全额退; 3-差额退',
    order_no              STRING COMMENT '订单号',
    refund_apply_type     BIGINT COMMENT '退款申请类型：1-仅退款；2-退货退款',
    refund_delivery       DECIMAL(27, 2) COMMENT '运费退款',
    sync_erp_status       BIGINT COMMENT '同步erp状态：-1-失败，0-未同步，1-成功',
    sync_erp_msg          STRING COMMENT '同步erp失败消息',
    create_sys_user_id    BIGINT COMMENT '操作人id',
    create_sys_user_name  STRING COMMENT '操作人名称',
    store_no              STRING COMMENT '门店编码',
    store_leader_id       BIGINT COMMENT '团长id'
)
COMMENT '订单退款表'
row format delimited fields terminated by ','
stored as orc
tblproperties ('orc.compress'='ZLIB');

2- 通过DataX完成数据导入操作
- 设置数据源: mysql连接 sale库 HIVE连接ODS库

在这里插入图片描述

配置DataX任务模板(后续调度时间依然凌晨20分, 可以省略配置)
构建任务

在这里插入图片描述

点击构建, 生成Json采集信息内容, 将其中append 更改为truncate
点击选择模板 –> 下一步
执行任务:

在这里插入图片描述

校验是否ok:

在这里插入图片描述

将临时表的数据导入到目标表

-- 开启非严格模式
set hive.exec.dynamic.partition.mode=nonstrict;

-- 从临时表中查询数据插入到真实表中
-- 方式1
insert overwrite table ods.ods_sale_shop_refund_i partition (dt)
select
    *,
   date_format(create_time,'yyyy-MM-dd') as dt
from ods.ods_sale_shop_refund_i_temp;

-- 方式2
insert overwrite table ods.ods_sale_shop_refund_i partition (dt)
select *,
	date(create_time) as dt
from ods.ods_sale_shop_refund_i_tmp;

在这里插入图片描述

删除临时表

drop table ods.ods_sale_shop_refund_i_temp;

增量导入: T+1模式

为了测试方便, 将MySQL中数据, 更为其中一条为上一天的数据

insert into sale.shop_refund (id, refund_no, refund_status, refund_code, refund_msg, refund_desc, create_time, update_time,
                         cancel_time, refund_amount, refund_point_amount, return_pay_point, return_point_amount,
                         refund_time, less_weight, pick_weight, is_deleted, refund_type, order_no, refund_apply_type,
                         refund_delivery, sync_erp_status, sync_erp_msg, create_sys_user_id, create_sys_user_name,
                         store_no, store_leader_id)
values (13,'220731Y28899211127',2,999,'其他原因','','2023-11-20 17:50:24','2023-11-20  17:50:24',null,5.26,5.00,null,null,'2023-11-20 17:50:23',null,null,0,1,'BL22073199620677',1,0.00,1,null,1001107,1001107,'Y288',null);

目标表是一个分区表, 需要将数据导入到上一天的分区中, 每天导入上一天的即可, 每次新增数据, 都是简历一个新的分区, 将上一天的数据放置到这个分区中

在这里插入图片描述

-- 如何获取上一天增量的sql语句?
date_format(create_time,'%Y-%m-%d') = DATE_FORMAT(date_sub(NOW(),INTERVAL 1 DAY),'%Y-%m-%d')

--或者:
create_time between concat(date_sub(current_date,INTERVAL 1 DAY),' 00:00:00') and concat(date_sub(current_date,INTERVAL 1 DAY),' 23:59:59')

在这里插入图片描述

点击构建 –> 修改为truncate –> 点击选择模板 –> 下一步生成任务

修改全局参数: dt的分区

在这里插入图片描述

执行任务

在这里插入图片描述

通过hive查询:
在这里插入图片描述

会发现压根没有数据, 但是数据确实在HDFS中存在了

查看当前有多少个分区内容?

在这里插入图片描述

思考: 为啥没有呢?　分区数据从某种角度来说, 应该是属于元数据的, 而DataX在导入的时候, 并不会生成这个分区元数据信息, 所以导致没有此分区, 也就无法映射到这部分的数据了

如何解决呢? 
	方式一: 手动添加一个分区即可
	alter table ods.ods_sale_shop_refund_i add partition (dt='2023-06-18')
	
	方式二: HIVE自动修复分区 (比较适合于有多个分区的情况)
	 MSCK REPAIR TABLE ods.ods_sale_shop_refund_i;
	

在DataX中如何解决呢?  在写出到HIVE中, 添加一个后置SQL: 
MSCK REPAIR TABLE ods.ods_sale_shop_refund_i

在这里插入图片描述

添加如下:

"postSql":["MSCK REPAIR TABLE ods.ods_sale_shop_refund_i"],

HIVE的函数库: https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF

2.3.3 新增及更新导入

以shop_order(商城订单表)

首次导入:

与仅新增的首次导入流程方案一模一样

增量导入:

--筛选条件
create_time between concat(date_sub(current_date,INTERVAL 1 DAY),' 00:00:00') and concat(date_sub(current_date,INTERVAL 1 DAY),' 23:59:59') 
OR 
last_update_time between concat(date_sub(current_date,INTERVAL 1 DAY),' 00:00:00') and concat(date_sub(current_date,INTERVAL 1 DAY),' 23:59:59')

--或者:
date_format(create_time,'%Y-%m-%d') = DATE_FORMAT(date_sub(NOW(),INTERVAL 1 DAY),'%Y-%m-%d')
OR 
date_format(last_update_time,'%Y-%m-%d') = DATE_FORMAT(date_sub(NOW(),INTERVAL 1 DAY),'%Y-%m-%d')

2.3.4 全量同步

以:门店商品库存天表为例

这里跟全量覆盖表的区别是：ods表是一个分区表，即每次同步全量数据到一个新的分区里，不在进行覆盖

ale_shop_refund_i


[外链图片转存中...(img-3DerZWCP-1700661476536)]

[外链图片转存中...(img-mW8Udnbg-1700661476537)]



添加如下:

```properties
"postSql":["MSCK REPAIR TABLE ods.ods_sale_shop_refund_i"],

HIVE的函数库: https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF

2.3.3 新增及更新导入

以shop_order(商城订单表)

首次导入:

与仅新增的首次导入流程方案一模一样

增量导入:

--筛选条件
create_time between concat(date_sub(current_date,INTERVAL 1 DAY),' 00:00:00') and concat(date_sub(current_date,INTERVAL 1 DAY),' 23:59:59') 
OR 
last_update_time between concat(date_sub(current_date,INTERVAL 1 DAY),' 00:00:00') and concat(date_sub(current_date,INTERVAL 1 DAY),' 23:59:59')

--或者:
date_format(create_time,'%Y-%m-%d') = DATE_FORMAT(date_sub(NOW(),INTERVAL 1 DAY),'%Y-%m-%d')
OR 
date_format(last_update_time,'%Y-%m-%d') = DATE_FORMAT(date_sub(NOW(),INTERVAL 1 DAY),'%Y-%m-%d')