OpenMLDB 发布线上到线下数据自动同步工具

2023-07-20
新加坡
本文字数：3107 字
阅读完需：约 10 分钟

OpenMLDB 正式发布线上到线下数据自动同步工具

OpenMLDB 作为一个线上线下一致的实时特征计算平台，通过合理的线上线下数据存储，确保线上到线下数据的一致性。由于线上和线下数据有不同性能和数据量的需求，通常情况下，OpenMLDB 的线上和线下的数据在物理上是分开存储。

在以前版本中，需要用户自行维护线上线下数据的同步和一致性，带来了一定的维护复杂度。近期，OpenMLDB v0.8.0 正式推出自动化线上到线下数据同步工具，实现了从实时数据库到离线数仓的自动同步，改进了手动维护的运维复杂度。下表总结了两种不同使用方式的优劣势。

线上到线下数据存储自动化同步工具的架构如下图所示，用户只需要将新的数据写入 OpenMLDB 线上实时数据库，并设置好线上到线下的同步机制，OpenMLDB 即可自动化地将数据实时或者定时地同步至一个或者多个离线数仓。OpenMLDB 的实时数据库根据数据过期机制仅保存用于线上特征计算的数据，而离线数仓将保留所有全量数据。

线上到线下自动同步实战演示

使用线上到线下自动同步功能，需要在每台 TabletServer 所在机器上部署至少一台 DataCollector，用于收集在线数据。接收并写入离线存储的工具 SyncTool 可放于任何机器，目前仅支持单体运行。详细部署方式见最新产品相关文档（https://openmldb.ai/docs/zh/main/tutorial/online_offline_sync.html ）。注意，目前在线存储仅支持磁盘表，离线支持写入到 HDFS。接下来将以 Docker 镜像为例，展示在离线同步方式。主要分为以下四步：

HDFS 环境配置，作为离线存储地址
OpenMLDB 部署，包含同步组件
创建 OpenMLDB 线上到线下的同步任务
在线数据导入，并检查离线存储地址

步骤一：HDFS 环境配置

HADOOP 安装参考 Pseudo-Distributed Operation，需配置 JAVA_HOME 和 PDSH_RCMD_TYPE 环境变量。为了简单起见，额外配置所有 USER 为当前用户 root。因此安装步骤为:

curl -SLO https://archive.apache.org/dist/hadoop/common/hadoop-3.2.2/hadoop-3.2.2.tar.gztar zxf hadoop-3.2.2.tar.gzapt update && apt install ssh pdsh service ssh startssh-keygen -t rsa -P '' -f ~/.ssh/id_rsacat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keyschmod 0600 ~/.ssh/authorized_keysecho -e "export HDFS_NAMENODE_USER=root\nexport HDFS_DATANODE_USER=root\nexport HDFS_SECONDARYNAMENODE_USER=root\nexport YARN_RESOURCEMANAGER_USER=root\nexport YARN_NODEMANAGER_USER=root\nexport JAVA_HOME=/usr/local/openjdk-11\nexport PDSH_RCMD_TYPE=ssh\n" >> hadoop-3.2.2/etc/hadoop/hadoop-env.shbash hadoop-3.2.2/sbin/start-dfs.sh

复制代码

可以通过ps axu检查，应该存在 3 个 HADOOP 相关的进程。

步骤二：OpenMLDB 部署

OpenMLDB 的部署使用/work/init.sh，配置离线可以读取 HDFS 集群，离线表默认存储地址可不更改，后面可以软链接的方式加载离线地址。虽然有两个 TabletServer，但实际只有一台机器，所以 DataCollector 只需要一个。注意 init.sh 启动的 TabletServer IP 为 localhost，我们需要将 DataCollector 的 IP 也改为 localhost，端口使用默认端口。启动 DataCollector 后再启动 SyncTool，SyncTool 需要注意 HADOOP 配置的读取，这里直接设置读取 HADOOP 安装目录中的配置文件。

export HADOOP_CONF_DIR=/work/hadoop-3.2.2/etc/hadoop//work/init.shcd /work/openmldbsed -i'' 's/endpoint=127.0.0.1/endpoint=localhost/g' conf/data_collector.flagsbash bin/start.sh start data_collectorsed -i'' 's#hadoop.conf.dir=.*#hadoop.conf.dir=/work/hadoop-3.2.2/etc/hadoop/#g' conf/synctool.propertiesbash bin/start.sh start synctool

复制代码

步骤三：创建同步任务

在 OpenMLDB 中创建磁盘表，才能进行同步任务的创建。参考OpenMLDB快速上手，创建一个磁盘表。

CREATE DATABASE demo_db;USE demo_db;CREATE TABLE demo_table1(c1 string, c2 int, c3 bigint, c4 float, c5 double, c6 timestamp, c7 date, INDEX(KEY=c1, TS=c6)) OPTIONS (storage_mode='HDD');

复制代码

使用tools/synctool_helper.py来管理同步任务，由于 SyncTool 使用默认 IP，所以以下命令均可以不填写-s。

创建同步任务，可以对某一张表创建同步任务，并且有三种同步模式可以选择，同时还需要填写 SyncTool 地址和离线数据地址。其中，离线数据地址指 SyncTool 可读取的 HDFS 地址，所以，需要 SyncTool 处有 HDFS 配置，地址不需要hdfs://路径开头。如果存入hdfs://<cluster>/tmp/hdfs-dest则填写/tmp/hdfs-dest。

# mode 0: 全量同步，在某一时刻无新数据读取时，将停止同步python3 tools/synctool_helper.py create -t db.table -m 0 -s <sync tool endpoint> -d /tmp/hdfs-dest # mode 1: 持续同步，同步任务将一直存在，并且对所有数据进行ts的筛选，例如-t时间为2023/7/18 00:00:00.000，则所有早于此的数据将不会被同步到离线python3 tools/synctool_helper.py create -t db.table -m 1 -ts 233 -s <sync tool endpoint> -d /tmp/hdfs-dest # mode 2: 持续同步，同步任务将一直存在，所有在线数据将同步到离线python3 tools/synctool_helper.py create -t db.table -m 2 -s <sync tool endpoint> -d /tmp/hdfs-dest

复制代码

在容器中，创建这样一个同步任务：

python3 tools/synctool_helper.py create -t demo_db.demo_table1 -m 2 -d /tmp/hdfs-dest

复制代码

步骤四：同步任务测试和验证

同步任务创建后，可以使用以下命令查询任务状态。

python3 tools/synctool_helper.py status

复制代码

可以看到该表的所有分区同步任务，也可以看到它们是否处于正常运行状态。

8 tasks(tid-pid){'lastUpdateTime': '2023-07-18 07:01:07.589000', 'progressPath': '/tmp/sync_task_progress/7/0.progress', 'tid': 7, 'pid': 0, 'mode': 'kFullAndContinuous', 'startTs': '0', 'syncPoint': {'type': 'kBINLOG', 'offset': '1'}, 'tabletEndpoint': 'localhost:10922', 'desEndpoint': '0.0.0.0:8848', 'token': '3e8e9997-5608-4106-bfc3-6923ac0b6939', 'dest': '/tmp/hdfs-dest', 'dataCollector': 'localhost:8888', 'count': '0', 'status': 'RUNNING'}... // other 7 taskstable scoperunning tasks:tid 7 task is running, pid parts: dict_keys([0, 1, 2, 3, 4, 5, 6, 7])all tasks are running

复制代码

此时 HADOOP 上还没有数据，仅创建了目录，可以通过以下方式查询。

/work/hadoop-3.2.2/bin/hdfs dfs -ls /tmp/hdfs-dest

复制代码

在线数据导入

SET @@execute_mode='online';LOAD DATA INFILE 'file:///work/taxi-trip/data/data.parquet' INTO TABLE demo_table1 options(format='parquet', mode='append');

复制代码

离线结果检查

可以用 dfs 的方式检查数据是否同步，可以得到在线数据的所有数据，共 10 行。

/work/hadoop-3.2.2/bin/hdfs dfs -cat /tmp/hdfs-dest/* | wc -l

复制代码

更常见的方式，可以将此地址作为离线数据软链接，使离线端可以使用数据。

USE demo_db;LOAD DATA INFILE 'hdfs:///tmp/hdfs-dest' INTO TABLE demo_table1 options(format='csv', header=false, deep_copy=false, mode='append');DESC demo_table1;SET @@sync_job=true;SELECT * FROM demo_table1;

复制代码

因为同步任务是持续的，再进行在线LOAD DATA或INSERT，数据将持续同步到离线地址，直到同步任务被删除。注意，磁盘表如果索引相同，会进行覆盖（更多信息参考磁盘表），不会增加数据行数，所以不要重复导入同样的数据进行测试，同步工具将追加数据到离线，不会进行覆盖。