【大数据开发运维解决方案】Sqoop增量同步Oracle数据到hive:merge-key再次详解

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介: 这篇文章是基于上面连接的文章继续做的拓展,上篇文章结尾说了如果一个表很大。我第一次初始化一部分最新的数据到hive表,如果没初始化进来的历史数据今天发生了变更,那merge-key的增量方式会不会报错呢?之所以会提出这个问题,是因为笔者真的有这个测试需求,接下来先对oracle端的库表数据做下修改,来模拟这种场景。

前言

对于sqoop增量同步Oracle数据到hive的命令参数以及如何定制自动增量job的测试已经再前面几篇文章详细测试过了,这篇文章是基于上面连接的文章继续做的拓展,上篇文章结尾说了如果一个表很大。我第一次初始化一部分最新的数据到hive表,如果没初始化进来的历史数据今天发生了变更,那merge-key的增量方式会不会报错呢?之所以会提出这个问题,是因为笔者真的有这个测试需求,接下来先对oracle端的库表数据做下修改,来模拟这种场景。


一、先插入一条数据

当前时间为:

SQL> select sysdate from dual;

SYSDATE
-------------------
2019-03-25 18:20:26

为了模拟我是有一部分历史数据没有导入到hive表,我这里先给oracle表插入一条历史数据:

SQL> select * from inr_job;

     EMPNO ENAME      JOB           SAL ETLTIME
---------- ---------- --------- ---------- -------------------
     1 er          CLERK           800 2019-03-22 17:24:42
     2 ALLEN      SALESMAN          1600 2019-03-22 17:24:42
     3 WARD       SALESMAN          1250 2019-03-22 17:24:42
     4 JONES      MANAGER          2975 2019-03-22 17:24:42
     5 MARTIN     SALESMAN          1250 2019-03-22 17:24:42
     6 zhao       DBA          1000 2019-03-22 17:24:42
     7 yan          BI           100 2019-03-22 17:24:42
     8 dong       JAVA           400 2019-03-22 17:24:42

8 rows selected.


SQL> insert into inr_job values(9,'test','test',200,sysdate-20);

1 row created.

SQL> commit;

Commit complete.

SQL> select * from inr_job;

     EMPNO ENAME      JOB           SAL ETLTIME
---------- ---------- --------- ---------- -------------------
     1 er          CLERK           800 2019-03-22 17:24:42
     2 ALLEN      SALESMAN          1600 2019-03-22 17:24:42
     3 WARD       SALESMAN          1250 2019-03-22 17:24:42
     4 JONES      MANAGER          2975 2019-03-22 17:24:42
     5 MARTIN     SALESMAN          1250 2019-03-22 17:24:42
     6 zhao       DBA          1000 2019-03-22 17:24:42
     7 yan          BI           100 2019-03-22 17:24:42
     8 dong       JAVA           400 2019-03-22 17:24:42
     9 test       test           200 2019-03-05 18:53:23--模仿没初始化到hive表的his数据

9 rows selected.

二、更新历史数据

接下来手动更新一下这个历史数据

SQL> update inr_job set sal=999,etltime=sysdate where empno=9;

1 row updated.

SQL> commit;

Commit complete.

查询一下表数据

SQL> select * from inr_job;

     EMPNO ENAME      JOB           SAL ETLTIME
---------- ---------- --------- ---------- -------------------
     1 er          CLERK           800 2019-03-22 17:24:42
     2 ALLEN      SALESMAN          1600 2019-03-22 17:24:42
     3 WARD       SALESMAN          1250 2019-03-22 17:24:42
     4 JONES      MANAGER          2975 2019-03-22 17:24:42
     5 MARTIN     SALESMAN          1250 2019-03-22 17:24:42
     6 zhao       DBA          1000 2019-03-22 17:24:42
     7 yan          BI           100 2019-03-22 17:24:42
     8 dong       JAVA           400 2019-03-22 17:24:42
     9 test       test           999 2019-03-25 18:54:39

9 rows selected.

现在数据发生了变动,然后去执行一下增量脚本:

[root@hadoop hadoop]# sqoop job --exec auto_job
Warning: /hadoop/sqoop/../accumulo does not exist! Accumulo imports will fail.
Please set $ACCUMULO_HOME to the root of your Accumulo installation.
19/03/25 18:55:49 INFO sqoop.Sqoop: Running Sqoop version: 1.4.7
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/hadoop/hbase/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/hadoop/hive/lib/log4j-slf4j-impl-2.6.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
19/03/25 18:55:51 INFO oracle.OraOopManagerFactory: Data Connector for Oracle and Hadoop is disabled.
19/03/25 18:55:51 INFO manager.SqlManager: Using default fetchSize of 1000
19/03/25 18:55:51 INFO tool.CodeGenTool: Beginning code generation
19/03/25 18:55:52 INFO manager.OracleManager: Time zone has been set to GMT
19/03/25 18:55:52 INFO manager.SqlManager: Executing SQL statement: SELECT t.* FROM INR_JOB t WHERE 1=0
19/03/25 18:55:52 INFO orm.CompilationManager: HADOOP_MAPRED_HOME is /hadoop
Note: /tmp/sqoop-root/compile/f64e34273a58459369885b96fe46a1ad/INR_JOB.java uses or overrides a deprecated API.
Note: Recompile with -Xlint:deprecation for details.
19/03/25 18:55:56 INFO orm.CompilationManager: Writing jar file: /tmp/sqoop-root/compile/f64e34273a58459369885b96fe46a1ad/INR_JOB.jar
19/03/25 18:55:56 INFO manager.OracleManager: Time zone has been set to GMT
19/03/25 18:55:56 INFO manager.SqlManager: Executing SQL statement: SELECT t.* FROM INR_JOB t WHERE 1=0
19/03/25 18:55:56 INFO tool.ImportTool: Incremental import based on column ETLTIME
19/03/25 18:55:56 INFO tool.ImportTool: Lower bound value: TO_TIMESTAMP('2019-03-25 18:50:07.0', 'YYYY-MM-DD HH24:MI:SS.FF')
19/03/25 18:55:56 INFO tool.ImportTool: Upper bound value: TO_TIMESTAMP('2019-03-25 18:55:56.0', 'YYYY-MM-DD HH24:MI:SS.FF')
19/03/25 18:55:56 INFO manager.OracleManager: Time zone has been set to GMT
19/03/25 18:55:56 INFO mapreduce.ImportJobBase: Beginning import of INR_JOB
19/03/25 18:55:56 INFO Configuration.deprecation: mapred.jar is deprecated. Instead, use mapreduce.job.jar
19/03/25 18:55:56 INFO manager.OracleManager: Time zone has been set to GMT
19/03/25 18:55:56 INFO Configuration.deprecation: mapred.map.tasks is deprecated. Instead, use mapreduce.job.maps
19/03/25 18:55:56 INFO client.RMProxy: Connecting to ResourceManager at /192.168.1.66:8032
19/03/25 18:55:59 INFO db.DBInputFormat: Using read commited transaction isolation
19/03/25 18:55:59 INFO mapreduce.JobSubmitter: number of splits:1
19/03/25 18:56:00 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1553503985304_0013
19/03/25 18:56:00 INFO impl.YarnClientImpl: Submitted application application_1553503985304_0013
19/03/25 18:56:00 INFO mapreduce.Job: The url to track the job: http://hadoop:8088/proxy/application_1553503985304_0013/
19/03/25 18:56:00 INFO mapreduce.Job: Running job: job_1553503985304_0013
19/03/25 18:56:10 INFO mapreduce.Job: Job job_1553503985304_0013 running in uber mode : false
19/03/25 18:56:10 INFO mapreduce.Job:  map 0% reduce 0%
19/03/25 18:56:19 INFO mapreduce.Job:  map 100% reduce 0%
19/03/25 18:56:20 INFO mapreduce.Job: Job job_1553503985304_0013 completed successfully
19/03/25 18:56:20 INFO mapreduce.Job: Counters: 30
    File System Counters
        FILE: Number of bytes read=0
        FILE: Number of bytes written=144777
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=87
        HDFS: Number of bytes written=38
        HDFS: Number of read operations=4
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=2
    Job Counters 
        Launched map tasks=1
        Other local map tasks=1
        Total time spent by all maps in occupied slots (ms)=5870
        Total time spent by all reduces in occupied slots (ms)=0
        Total time spent by all map tasks (ms)=5870
        Total vcore-milliseconds taken by all map tasks=5870
        Total megabyte-milliseconds taken by all map tasks=6010880
    Map-Reduce Framework
        Map input records=1
        Map output records=1
        Input split bytes=87
        Spilled Records=0
        Failed Shuffles=0
        Merged Map outputs=0
        GC time elapsed (ms)=100
        CPU time spent (ms)=3220
        Physical memory (bytes) snapshot=189059072
        Virtual memory (bytes) snapshot=2147303424
        Total committed heap usage (bytes)=102236160
    File Input Format Counters 
        Bytes Read=0
    File Output Format Counters 
        Bytes Written=38
19/03/25 18:56:20 INFO mapreduce.ImportJobBase: Transferred 38 bytes in 23.7426 seconds (1.6005 bytes/sec)
19/03/25 18:56:20 INFO mapreduce.ImportJobBase: Retrieved 1 records.
19/03/25 18:56:20 INFO tool.ImportTool: Final destination exists, will run merge job.
19/03/25 18:56:20 INFO Configuration.deprecation: mapred.output.key.class is deprecated. Instead, use mapreduce.job.output.key.class
19/03/25 18:56:20 INFO client.RMProxy: Connecting to ResourceManager at /192.168.1.66:8032
19/03/25 18:56:22 INFO input.FileInputFormat: Total input paths to process : 2
19/03/25 18:56:23 INFO mapreduce.JobSubmitter: number of splits:2
19/03/25 18:56:23 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1553503985304_0014
19/03/25 18:56:23 INFO impl.YarnClientImpl: Submitted application application_1553503985304_0014
19/03/25 18:56:23 INFO mapreduce.Job: The url to track the job: http://hadoop:8088/proxy/application_1553503985304_0014/
19/03/25 18:56:23 INFO mapreduce.Job: Running job: job_1553503985304_0014
19/03/25 18:56:37 INFO mapreduce.Job: Job job_1553503985304_0014 running in uber mode : false
19/03/25 18:56:37 INFO mapreduce.Job:  map 0% reduce 0%
19/03/25 18:56:46 INFO mapreduce.Job:  map 100% reduce 0%
19/03/25 18:56:56 INFO mapreduce.Job:  map 100% reduce 100%
19/03/25 18:56:57 INFO mapreduce.Job: Job job_1553503985304_0014 completed successfully
19/03/25 18:56:57 INFO mapreduce.Job: Counters: 49
    File System Counters
        FILE: Number of bytes read=614
        FILE: Number of bytes written=435819
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=657
        HDFS: Number of bytes written=361
        HDFS: Number of read operations=9
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=2
    Job Counters 
        Launched map tasks=2
        Launched reduce tasks=1
        Data-local map tasks=2
        Total time spent by all maps in occupied slots (ms)=11103
        Total time spent by all reduces in occupied slots (ms)=7376
        Total time spent by all map tasks (ms)=11103
        Total time spent by all reduce tasks (ms)=7376
        Total vcore-milliseconds taken by all map tasks=11103
        Total vcore-milliseconds taken by all reduce tasks=7376
        Total megabyte-milliseconds taken by all map tasks=11369472
        Total megabyte-milliseconds taken by all reduce tasks=7553024
    Map-Reduce Framework
        Map input records=9
        Map output records=9
        Map output bytes=590
        Map output materialized bytes=620
        Input split bytes=296
        Combine input records=0
        Combine output records=0
        Reduce input groups=9
        Reduce shuffle bytes=620
        Reduce input records=9
        Reduce output records=9
        Spilled Records=18
        Shuffled Maps =2
        Failed Shuffles=0
        Merged Map outputs=2
        GC time elapsed (ms)=263
        CPU time spent (ms)=3980
        Physical memory (bytes) snapshot=670138368
        Virtual memory (bytes) snapshot=6394978304
        Total committed heap usage (bytes)=508559360
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=361
    File Output Format Counters 
        Bytes Written=361
19/03/25 18:56:57 INFO tool.ImportTool: Saving incremental import state to the metastore
19/03/25 18:56:57 INFO tool.ImportTool: Updated data for job: auto_job

发现没有报错唉,然后去看看hive表:

hive> select * from inr_job;
OK
1    er    CLERK    800.0    2019-03-22 17:24:42.0
2    ALLEN    SALESMAN    1600.0    2019-03-22 17:24:42.0
3    WARD    SALESMAN    1250.0    2019-03-22 17:24:42.0
4    JONES    MANAGER    2975.0    2019-03-22 17:24:42.0
5    MARTIN    SALESMAN    1250.0    2019-03-22 17:24:42.0
6    zhao    DBA    1000.0    2019-03-22 17:24:42.0
7    yan    BI    100.0    2019-03-22 17:24:42.0
8    dong    JAVA    400.0    2019-03-22 17:24:42.0
9    test    test    999.0    2019-03-25 18:54:39.0
Time taken: 0.336 seconds, Fetched: 9 row(s)

没初始化进来的历史数据在近期变动之后,如果符合增量条件的话,也会append进来并不会报错,完全符合笔者需求,其实看看merge-key参数大致原理,也是知道这样是可行的,毕竟是通过主键和最后修改时间去做增量合并。


总结

上面是博主对merge-key做的再一次深入测试,因为实际工作中的确会用到,那就自己研究清楚了!

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps 
相关文章
|
22天前
|
分布式计算 大数据 关系型数据库
MaxCompute产品使用问题之如何实现MySQL的实时增量同步
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
22天前
|
分布式计算 DataWorks 大数据
MaxCompute产品使用问题之DataWorks整库全增量同步任务的源库如果新增了表,如何能将这个表快速同步进maxcompute
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
29天前
|
SQL 分布式计算 HIVE
实时计算 Flink版产品使用问题之同步到Hudi的数据是否可以被Hive或Spark直接读取
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
1月前
|
SQL 数据采集 数据可视化
基于Hive的招聘网站的大数据分析系统
基于Hive的招聘网站的大数据分析系统
|
1月前
|
SQL 分布式计算 NoSQL
使用Spark高效将数据从Hive写入Redis (功能最全)
使用Spark高效将数据从Hive写入Redis (功能最全)
|
1月前
|
SQL 关系型数据库 MySQL
基于Hive的天气情况大数据分析系统(通过hive进行大数据分析将分析的数据通过sqoop导入到mysql,通过Django基于mysql的数据做可视化)
基于Hive的天气情况大数据分析系统(通过hive进行大数据分析将分析的数据通过sqoop导入到mysql,通过Django基于mysql的数据做可视化)
|
2月前
|
SQL 分布式计算 关系型数据库
使用 Spark 抽取 MySQL 数据到 Hive 时某列字段值出现异常(字段错位)
在 MySQL 的 `order_info` 表中,包含 `order_id` 等5个字段,主要存储订单信息。执行按 `create_time` 降序的查询,显示了部分结果。在 Hive 中复制此表结构时,所有字段除 `order_id` 外设为 `string` 类型,并添加了 `etl_date` 分区字段。然而,由于使用逗号作为字段分隔符,当 `address` 字段含逗号时,数据写入 Hive 出现错位,导致 `create_time` 值变为中文字符串。问题解决方法包括更换字段分隔符或使用 Hive 默认分隔符 `\u0001`。此案例提醒在建表时需谨慎选择字段分隔符。
|
2月前
|
SQL 存储 JSON
Hive 解析 JSON 字符串数据的实现方式
Hive 提供 `get_json_object` 函数解析 JSON 字符串,如 `{"database":"maxwell"}`。`path` 参数使用 `$`、`.`、`[]` 和 `*` 来提取数据。示例中展示了如何解析复杂 JSON 并存储到表中。此外,Hive 3.0.0及以上版本内置 `JsonSerDe` 支持直接处理 JSON 文件,无需手动解析。创建表时指定 `JsonSerDe` 序列化器,并在 HDFS 上存放 JSON 文件,可以直接查询字段内容,方便快捷。
|
29天前
|
消息中间件 存储 SQL
实时计算 Flink版产品使用问题之kafka2hive同步数据时,如何回溯历史数据
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
15天前
|
运维 监控 测试技术
自动化运维实践:CI/CD流程详解
【6月更文挑战第30天】CI/CD实践推动软件开发自动化,通过持续集成确保代码质量,自动部署提升交付速度。核心流程包括:代码管理(Git等)、自动化构建与测试、代码审查、部署。关键点涉及选择工具、测试覆盖率、监控及团队协作。采用CI/CD能减少错误,但需应对挑战,如工具选型、全面测试和团队沟通。

推荐镜像

更多