(19)国家知识产权局 (12)发明 专利申请 (10)申请公布号 (43)申请公布日 (21)申请 号 202210823857.8 (22)申请日 2022.07.14 (71)申请人 广东嘉贸通科技有限公司 地址 510000 广东省广州市海珠区滨江西 路26-28号602房北侧 (72)发明人 钟振帆　 (74)专利代理 机构 广州越华专利代理事务所 (普通合伙) 44523 专利代理师 陈岑 (51)Int.Cl. G06F 16/215(2019.01) G06F 16/23(2019.01) G06F 16/27(2019.01) (54)发明名称 基于海关业 务大数据的实时数据采集方法 (57)摘要 本发明公开了一种基于海关业务大数据的 实时数据采集方法， 涉及海关大数据分析处理技 术领域， 其包括： 接收和获取用户配置的采集任 务； 根据采集任务中的OGG采集对象， 下 发配置到 OGG采集程序； 创建OGG ‑KafKa专用T opic， 并配置 相关Topic备份策略， 将已建好的专用T opic名称 写入到OGG的配置文件； 以及， 使用KafKa  sink‑ connector组件构建数据回写任 务， 将OGG ‑Topic 中采集后的数据回写到Hive。 本申请具有缓解数 据时效性与计算效率相矛盾的问题， 兼顾优点的 效果。 权利要求书1页 说明书4页 附图1页 CN 115098486 A 2022.09.23 CN 115098486 A 1.一种基于海关业 务大数据的实时数据采集方法， 其特 征在于， 包括以下步骤： 接收和获取用户配置的采集任务； 根据采集任务中的O GG采集对象， 下发配置 到OGG采集程序； 创建OGG‑KafKa专用Topic， 并配置相关Topic备份策略， 将已建好的专用Topic名称写 入到OGG的配置文件； 以及， 使用KafKa  sink‑connector组件构建数据回写任务， 将OGG ‑Topic中采集后的数据回 写到Hive； 其中， 当使用OGG采集程序的组件采集目标表时， 对数据源端与目标端的OGG  mgr进行 配置extract、  pump进程、 replicate进程、 define文件并执行defgen命令， 并将生成的prm 文件上传到目标端； 以及， 按照源mgr—目标端mgr—源extract—源pump—目标端replicate的顺序启动全部进 程， 完成O GG与KafKa 中OGG‑Topic的数据抽取任务 通道的搭建。 2.根据权利要求1所述的基于海关业务大数据的实时数据采集方法， 其特征在于： 从 KafKa的OGG ‑Topic中进行批量数据消费后， 对该批数据按照时间序列进行排序和数据清 洗。 3.根据权利要求2所述的基于海关业务大数据的实时数据采集方法， 其特征在于： 所述 对该批数据按照时间序列进行排序和数据清洗， 其包括： 若该批数据中相同主键最后的事务操作为delete操作， 则直接去除该主键相关的入库 记录， 并分发到delete操作队列； 若对相同主键最后的事务操作 为insert操作， 则只保留最后一条数据入库记录进行插 入操作， 并分发到i nsert操作队列； 若对相同主键进行连续多次的update事务操作， 则只保留最后一次的update操作记 录， 并分发到update操作队列； 三个队列获取到 Hive的增、 删、 改的数据操作记录分别以三张相应的临时表保存。 4.根据权利要求3所述的基于海关业务大数据的实时数据采集方法， 其特征在于： 所述 对该批数据按照时间序列进行排序和数据清洗， 其包括： 新增数据， 则通过insert  into语 句对数据表进行相应批量的新增。 5.根据权利要求3所述的基于海关业务大数据的实时数据采集方法， 其特征在于： 所述 对该批数据按照时间序列进行排序和数据 清洗， 其包括： 更新操作， 则需获取到更新字段的 唯一主键， 存 入缓存； 根据缓存中的ID去查询数据表里的数据与更新的数据合并， 存入临时表， 并执行merge   into更新到表数据。 6.根据权利要求3所述的基于海关业务大数据的实时数据采集方法， 其特征在于： 所述 对该批数据按照时间序列进行排序和数据清洗， 其包括： 删除操作， 则根据主键来执行 merge into的删除操作。权　利　要　求　书 1/1 页 2 CN 115098486 A 2基于海关业务大数据的实时数据采集方 法 技术领域 [0001]本申请涉及海关大数据分析处理技术领域， 尤其是涉及 一种基于海关业务大数据 的实时数据采集方法。 背景技术 [0002]目前， 海关大数据平台的实时数据采集方案是： 通过使用Spoop进行全量数据抽 取， 将数据源的数据抽取到预先建好的HBase表， 然后使用关系型数据库的实时归档日志数 据以及Oracle  GoldenGate （以下简称OGG） ， 将海关前台业务数据库的实时增量数据采集入 Hadoop平台的HBase 数据库。 [0003]上述的实时数据采集方案虽然能实现数据的时效性， 但数据采集进入HBase数据 库， 而非采集落地到Hadoop平台的HDFS文件体系， 带来的后续问题则是： 离线计算的性能受 到影响， 因此本申请提出一种新的方案 。 发明内容 [0004]为了缓解数据时效性与计算效率相矛盾， 兼顾两者， 本申请提供一种基于海关业 务大数据的实时数据采集方法。 [0005]本申请提供一种基于海关业务大数据的实时数据采集方法， 采用如下的技术方 案： 一种基于海关业 务大数据的实时数据采集方法， 包括以下步骤： 接收和获取用户配置的采集任务； 根据采集任务中的O GG采集对象， 下发配置 到OGG采集程序； 创建OGG‑KafKa专用Topic， 并配置相关Topic备份策略， 将已建好的专用Topic名 称写入到O GG的配置文件； 以及， 使用KafKa  sink‑connector组件构建数据回写任务， 将OGG ‑Topic中采集后的数 据回写到 Hive； 其中， 当使用OGG采集程序的组件采集目标表时， 对数据源端与目标端的OGG  mgr 进行配置 extract、  pump进程、 replicate进程、 define文件并执行defgen命令， 并将生成的 prm文件上传到目标端； 以及， 按照源mgr— 目标端mgr—源extr act—源pump— 目标端replicate的顺序启动全 部进程， 完成O GG与KafKa 中OGG‑Topic的数据抽取任务 通道的搭建。 [0006]可选的， 从KafKa的OGG ‑Topic中进行批量数据消费后， 对该批数据按照时间序列 进行排序和数据清洗 。 [0007]可选的， 所述对该批数据按照时间序列进行排序和数据清洗， 其包括： 若该批数据中相同主键最后的事务操作为delete操作， 则直接去除该主键相关的 入库记录， 并分发到delete操作队列； 若对相同主键最后的事务操作为insert操作， 则只保 留最后一条数据 入库记录进说　明　书 1/4 页 3 CN 115098486 A 3