乐趣区

关于java:Tapdata-开源项目基础教程功能特性及实操演示

jiezi

自开源以来，Tapdata 吸引了越来越多开发者的关注。随着更多陈腐力量涌入社区，在和社区成员探讨共创的过程中，咱们也意识到在根底文档之外，一个更“直观”、更具“互动性”的实际示范教程的重要性和必要性。为了辅助开发者更好地了解技术文档，真正实现疾速上手、深度参加，即刻开启实时数据新体验，咱们同步启动了 Tapdata 性能个性及操作演示系列教程。

以下，为本教程的第一弹内容——零根底疾速上手实际，粗疏分享了从源码编译和启动服务到如何新建数据源，再到如何做数据源之间的数据同步的启动部署及常见性能演示，次要工作包含：

  • MongoDB => MySQL 的实时同步工作，包含主动建表与实时数据同步
  • MongoDB => MySQL + 自定义函数 + 主动模型推演 + 建表的工作，能够依照自定义函数的逻辑主动推断表模型
  • MySQL=> PG + DDL 同步的工作，能够主动同步数据, 也能够主动同步表模型变更

一、我的项目简介

什么是 Tapdata?

Tapdata 开源我的项目的定位是一个实时数据服务平台，目前已上线的 1.0 版本外围笼罩实时数据同步、实时数据开发、Fluent ETL 等场景，具备全量和增量复制、异构数据库间的同步与转换，表级同步以及工作监控等能力。其工作机制次要蕴含以下四个环节的性能个性：
基于 CDC 的无侵入数据源实时采集

  • 异构数据模型主动推断与转换
  • 数据处理，流式计算，缓存存储一体架构
  • 一键将模型公布为数据服务的闭环能力

如上图所示，最左侧是包含数据库、数仓以及利用文件等在内的各种数据源。通过 Tapdata 主打的基于 CDC 的无侵入数据实时采集模块，可能将来自这些数据源的数据实时抓取过去；再通过一个异构数据模型的主动推断和转换，成为计算流中规范数据的一部分；继而通过一些数据处理、流式计算，配合我的项目自带的缓存存储，将数据按开发者的需要实现转换；最初通过数据公布能力，以 API 的模式出现，或是间接按需传入数据指标，例如数据库、利用，或是 Web 服务等，从而达到更快获取所需数据的目标。

和传统解决方案的比照

同样的场景下，传统解决方案比拟常见的形式是通过脚本轮询读取、自定义代码连接器，或是 Kettle/Canal/OGG 一类的开源适配器等形式，对来自于各类数据源的数据进行读取；再通过 Kafka/MQ 这样的音讯队列，或是 Spark/Flink 这样的计算引擎等形式进行数据的流转转换、开发荡涤，进行数据的流转转换、开发荡涤；最终通过本人写一些 API 接口逻辑将数据发送到指标终端。

这类计划的整个链路从原始读取到音讯队列，再到写 Stream 逻辑，再到计算层，再到输出层，存在一个十分大且非常明显的缺点，就是链路长、非实时、老本高、难保护。

而 Tapdata 要做的，就是与之绝对的一种疾速、实时、简略、易用的计划——将两头的全副过程工作，交由咱们来实现，这也是 Tapdata 的开源设计理念。

二、应用阐明

如何从源码构建并启动残缺的 Tapdata 服务？

从 0 开始构建和部署 Tapdata 开源版本，须要实现三个步骤：

  1. 环境筹备: Linux + Docker（以后版本仅反对 Linux 和 Docker 环境，基于非 Docker 的和非 Linux 的环境正在适配中，很快会和大家见面）
  2. 下载源码: git clone git@github.com:tapdata/tapdata.git && cd tapdata
  3. 通过一个命令，一键编译所有组件并启动服务: bash build/quick-dev.sh

另附代码构造解析及启动阐明：
代码库次要组成部分（目录）

  • assets：用于存储咱们的图片、logo 等动态资源
  • build：次要用于寄存咱们用来打包、编译或测试的脚本，不是框架的次要组成部分 connectors
  • common：数据源的通用对象以及类的定义
  • plugin-kit：供开发者应用，如果开发者想在框架上开发一个本人的数据源，就能够用到这里的一些 API 办法
  • tapshell：目前 Tapdata 的开源我的项目没有裸露 UI 界面，整个零碎的交互应用都是通过
  • tapshell 来运行的，是一个基于 Python 的交互式命令行工具
  • bin：咱们后续会在这里放一些相似于启动 / 进行的脚本
  • connectors：目前 Tapdata 反对的数据源都在这里：
    开源数据库 Apache Doris
    测试用数据库 Dummy
    ES/Kafka/MySQL/PG/OceanBase/RabbitMQ/RocketMQ……
  • dist：每次打包时的成品目录
  • engine：引擎，做计算时的一个次要仓库，能够用来做数据的读取、转换，以及聚合计算
  • manager：治理端，开发者的工作治理、日志上报、监控等信息都是通过这个组件来实现的。咱们也是通过 manager 来实现工作的长久化（engine 自身仅运行工作，不做长久化）
  • README：一个中文版，一个英文版

编译阐明：先把 plugin-kit 和 connectors common 编译完（编译出咱们根本依赖的对象），再实现 engine 引擎编译，持续实现 manager 编译后，整个服务就能够跑起来了。最初通过 build 里的 DockerFile 打包成一个 All-in-One 的镜像，咱们最终运行的是这个镜像。在运行之后，咱们还须要把 connectors 上面的所有数据源注册到零碎中去。

数据源注册：如果想要注册一个新的数据源，能够通过 accesstoken 来实现

如何动静注册数据源？

以 MongoDB 为例，具体演示如何动静地注册一个数据源，让零碎具备插件式数据集成的能力：

1. 动静注册一个开发实现的数据源插件

  • 列出零碎以后反对的数据源：show connectors
  • 注册 MongoDB 数据源插件
  • 列出零碎以后反对的数据源: show connectors

  • 新建一个 MongoDB 数据源 

    DataSource("mongodb", $name).uri($uri).save()
  • 校验并保留数据源

2. 新建所需的全副数据源, 校验并保留数据源 

DataSource("mysql",$name).host($host).port($port).username($username).password($password).db($db).save()
DataSource("postgresql",$name).host($host).port($port).username($username).password($password).db($db).schema($schema).save()

查看数据源配置, 检查表加载状况

三、性能示例及工作操作演示

提前准备的库（已确认可连通）：

  • MongoDB: 4.2，蕴含一张表：car
  • MySQL: 8.0.22，空库
  • PG: 12.6，空库

构建并运行一个 MongoDB => MySQL 的简略同步工作

* 任务说明：这个同步工作不蕴含任何简单解决，要求是将 car 表从 MongoDB 同步至 MySQL。

1. 新建表同步工作并运行 

p = Pipeline('mongo=>mysql')
p.readFrom(mongodb.car).writeTo(mysql.car)
p.start()

2. 查看工作运行状况：show jobs
3. 监听实时状态：monitor job mongo=>mysql
4. 查看运行日志：logs job mongo=>mysql
5. 在源与指标查看数据状况
6. 做一个变更，查看数据是否同步

工作实现状况：数据的增删改都能够实时同步过去，提早个别在几百毫秒以内。

构建并运行一个 MongoDB => MySQL 的带 UDF 的同步工作

* 任务说明：从 MongoDB 到 MySQL 的带 UDF（User Defined Function）的数据同步，依然是数据同步，但做了一些字段变换。

1. 定义 UDF（User Defined Function）：加一些字段，删除一个已存在的数据

2. 新建表同步工作，减少 UDF 节点并运行 

p = Pipeline('mongo=>mysql_with_udf')
p.readFrom(mongodb.car).processor(pp).writeTo(mysql.car_with_udf)
p.start()

* 补充阐明：与上一个工作的不同之处在于，这里在读写之间减少了一个处理器（processor），对 record 做了一些变换的同时，依照 UDF 定义增减了字段，变换了类型。

3. 监听实时状态：monitor job mongo=>mysql
4. 查看运行日志：logs job mongo=>mysql
5. 查看建表状况

* 工作实现状况：亮点在于，不仅实现了数据同步，还实现了表构造的主动创立。在同步过程中，可能主动建出解决完之后的表构造，无需像其余同步工具一样手动建表，应用更不便。

构建并运行一个 MySQ => PG 的反对 DDL 同步工作

* 任务说明：从 MySQL 到 PG 的 DDL 同步，对象是第一个工作中用到的 car 表。

1. 新建表同步工作，配置反对 DDL 并运行 

p = Pipeline('mysql=>pg_with_ddl')
p.readFrom(mysql.car_ddl, ddl=True).writeTo(pg.car_ddl)
p.start()

* 补充阐明：DDL 工作是默认不开启的，如果想要同步 DDL，能够通过执行 ddl=True 关上。

2. 监听实时状态：monitor job mongo=>mysql
3. 查看运行日志：logs job mongo=>mysql
4. 查看建表状况
5. 源端进行屡次 DDL，查看指标是否同步：

  • 源字段类型调整
  • 源字段重命名
  • 新增 / 删除字段

* 工作实现状况：实现实时同步，提早个别也会在几百毫秒以内。

至此，咱们的整个零碎中就同时有如上三个工作在跑，咱们能够随时：

  • 通过 logs job + 工作名来查看工作运行状况；
  • 通过 monitor job + 工作名来查看该工作的运行指标信息，包含 input/output/update 的数据量等；
  • 通过 show/use 来查看表信息、表构造

👉🏻GitHub 我的项目链接：https://www.github.com/tapdat…

如果您对咱们的我的项目感兴趣，欢送给 Tapdata【Star+Fork+Watch】三连击

以上，就是本次教程的核心内容回顾。还有更多无关 Tapdata 开源我的项目的问题亟待解答？曾经顺利度过“自学上手期”，想要观看更多根底之上的细节操作？欢送扫码增加 Tapdata 小姐姐（微信号：Tapdata2022），退出 Tapdata 社区群进行进一步交换探讨。

咱们期待与您共创一个优良的开源我的项目以及一个凋谢成熟的开源社区，见证实时数据的更多可能，下一期教程内容，或者就源自您提出的问题！

🎁 更多进群彩蛋

在 Tapdata 社区群，开源我的项目核心成员将为大家线上答疑，帮忙开发者们疾速理清困惑。与此同时，作为 Tapdata 社区沉闷用户，你还能够：

  • 取得 Tapdata 开源 Issue、需要的非凡优先级
  • 第一工夫播种社区最新资讯（包含但不限于开发计划、核心技术、业务场景等）
  • 参加流动、支付开源体验官老手工作、取得商务双肩包、潮牌 T 恤等更多好礼
  • 有机会受邀退出 Tapdata Committer Program，成为正式的 Tapdata Committer
  • 有机会直接参与并影响 Tapdata 的将来走向

👉🏻戳这里，观看教程残缺回放

退出移动版