适用模块

内部环境实数据传输实时同步任务的整体开发流程

具体说明

用户可以参照如下demo，完整的用数据传输模块的实时同步功能进行开发

使用示例

数据传输实时同步数据任务demo

目前很多内部用户在使用数据传输实时同步任务，该类任务由于是封装sloth的接口提交到sloth运行的任务。和常用的离线同步任务有一些区别，故很多用户在使用时不太熟练，这里对整个流程做一个简单的介绍。

1.登记kafka数据源

按照平常登记数据源的流程，登记kafka数据源

2.建流表

实时同步任务，用户一般需求为将kafka中的数据实时消费导入到hive数仓表中。 目前数据传输的实时同步任务只支持流表(kafka)-hdfs，后续需要手动load进hive表。后面会做介绍。

首先需要建流表（这里项目的db已经默认建好了），可以直接创建表，如没有权限，可以申请权限。或者让管理员或者负责人点查看权限添加权限。该处权限不受安全中心元数据中心权限控制。

新建表选择kafka数据源类型，选择对应的topic和序列化方式。

填写字段类型 注：在flink sql 中，有很多复杂字段。类似map arrary row等复杂字段，在建流表时无法下拉选择。但可以手动填入，也可以保存。例如。

例如kafka中的这种数据

3.配置传输任务

流表建成功后，此时任务类似离线同步任务的配置方法

选择流表和要导入的hdfs路径。

任务运维可以在这里进行启动和恢复。或者进入sloth界面查看具体情况（文档后面会介绍） 运行后，可以在对应的hdfs中找到文件，下载并打开分析数据大致正不正确 正确的话可以往hive表塞。 首先按照流表结构，建相同的hive表(离线开发新建离线表也没有hive复杂类型的)，故需要手动写sql在自助分析建表。 但是hive中没有arrary row。可以建为arrary struct

这里和上面流表的arrary row一样。这里要保证数据类型一致，避免后面读数有问题。 切如果导入hdfs为json格式，hive建表序列化方式要改成json格式，防止读数有问题

且location要手动指定成实时同步任务指定的hdfs路径。否则读不到数据文件。 至此一套实时数据导入hive数仓的开发流程结束。 如果需要每天写入可以自己在猛犸离线开发设置调度的sql节点，可以在sql节点自行设置alter table add partition，location指向每天实时写入的数据路径。至此开发完成。

再介绍一下如何通过sloth维护 前面说过传输实际是封装了sloth的接口，任务提交到sloth执行的，对于内部环境来说。实际是维护了一个sloth的项目，任务都提交到固定的这个sloth项目和队列上去运行。 内部集群即为这个项目

这里首先加入猛犸项目，然后添加实时管理员权限(控制所有实时任务)或实时开发（控制自己的实实时任务）。

进入该项目的实时计算平台

在实时开发中，实时同步任务名+猛犸项目名+集群标识。即为上面提交的实时同步任务。例如

再点击运维，即可查看日志，点击启停。或者跳转flink页面。

在猛犸数据传输中实际也是包装这些接口。包括一些告警配置。

作者：刘思伟 日期：2022-06-07
版本：v1
更新：暂无