Kafka 接入

Apache Kafka 是一个开源分布式事件流平台，它可以被当作是一个消息系统，读写流式数据，帮助发布、订阅消息；也可以用于编写可扩展的流处理应用程序，来应付实时响应的场景；还可以与数据库对接，安全的将流式数据存储在一个分布式，有副本备份，可容错的集群中。可以说，它所生产的“事件流”是信息数据传输的“中枢神经”。

如果你正打算用 Kafka 将数据写入 YMatrix 集群，请不要错过这篇文档。YMatrix 数据库支持 Kafka 无缝连接功能，可将 Kafka 数据持续自动导入到 YMatrix 表中，并支持图形化界面操作。目前接入的数据格式可以是 CSV 和 JSON。我们将以一个最简单的例子，来介绍如何用 YMatrix 管理平台接入 Kafka 数据。

1 准备工作

1.1 搭建 Kafka 环境

首先你需要搭建一个健康的 Kafka 环境，参照官方指南 Kafka Quickstart 即可。

1.2 创建 Kafka 主题

现在，你已经在服务器上搭建好了 Kafka 环境，你可以通过以下命令或其他方式进入。

$ cd packages/kafka_2.13-3.2.0

然后启动 Kafka 服务并创建一个测试主题（Topic），端口号默认为 9092。

$ bin/kafka-topics.sh --create --topic csv_test --bootstrap-server localhost:9092

1.3 写入测试数据

然后分条写入几个测试数据，以 ctrl-c 结束：

$ bin/kafka-console-producer.sh --topic csv_test --bootstrap-server localhost:9092
>1,Beijing,123.05,true,1651043583,1651043583123,1651043583123456
>2,Shanghai,112.95,true,1651043583,1651043583123,1651043583123456
>3,Shenzhen,100.0,false,1651043583,1651043583123,1651043583123456
>4,Guangxi,88.5,false,1651043583,1651043583123,1651043583123456
>^C

通过上面命令向新创建的 csv_test 写入了4条数据，分别为 4 条逗号分割的 CSV 行。准备工作完毕，下面将介绍如何创建 Kafka 数据流来导入数据。

2 创建 Kafka 数据流

在浏览器里输入 MatrixUI 所在机器的 IP（默认是主节点的 IP）、端口号：

http://<IP>:8240

成功登录之后进入主界面，界面分为进程视图和任务视图，进程视图是指已开启的 MatrixGate 进程，任务视图是指所有导入数据的任务。其中，一个 MatrixGate 进程可以包含多个任务。

点击“从 Kafka 中导入数据”。

选择 Kafka 数据源。首先以 CSV 文件为例说明步骤。

2.1 CSV 数据流接入

2.1.1 选择或新建目标表

首先，你可以选择已有表或新建表来作为接收 Kafka 数据的目标表。

2.1.1.1 选择已有表

在目标表栏选择已有表与数据导入模式。这里假设选择的表为 postgres 数据库 public 模式下的 test。选择后，下方会列出该表的表结构（字段名、类型、是否允许为 NULL、是否自增），供核对字段。

数据导入模式决定重复数据的处理方式，按目标表的约束情况选择：

更新插入 / UPSERT：将 UPDATE 和 INSERT 结合，实现一条 SQL 内自动插入或者更新：如果记录不存在则插入，如果存在则更新。目标表必须有不含 serial 类型字段的唯一约束，且映射时必须指定「设备 ID（唯一标识）」。
去重插入 / DEDUPLICATE：为保证数据唯一性，在数据插入时，若多条数据唯一约束字段的值相同，仅保留第一条。目标表同样必须有不含 serial 类型字段的唯一约束。
直接插入：不对数据进行处理，按照数据生产的顺序依次导入。

2.1.1.2 选择新建表

新建表，这里假设模式为 public，表名为 test1。

新建表时先选择表类型：

分区表：按时间列自动分区，适合持续写入的时序数据。字段映射时必须选择时间列。
非分区表：创建不分区的 MARS3 表，不需要选择时间列。

新建表不需要在这一步定义表结构，字段名与类型在第 3 步「映射」中通过自动映射或手动编辑确定。导入模式只支持「更新插入」和「直接插入」两种。

选择「分区表」时，下方会出现自动分区设置：

选择「非分区表」时不涉及自动分区：

建表成功，出现红色的新建标识。

为其选择数据导入模式及自动分区策略。此处分区策略默认采用 auto_partitioning。非分区表只选择数据导入模式，不涉及自动分区策略。

2.1.2 源数据

目标表选择完成后，第二步设置源数据的连接。你需要完成以下操作，才可以正确地进入下一步：

首先，在左侧菜单栏按照图中格式在地址框中输入 Kafka broker 地址及端口。如果有多个 URL，需要用逗号分割。
然后，连接 Kafka 所需验证方式，并验证 Kafka 中的用户名和密码。
最后，在右侧菜单栏勾选或搜索需要连接的 Kafka 主题。

地址与认证信息可以保存为「Kafka 配置」反复使用：填写完成后点击「保存为配置」，输入配置名称并保存；之后创建数据流时可直接选用已保存的配置，界面会显示「正在使用配置：<配置名称>」，需要临时改动时点「改用手工配置」。删除配置不会影响已经创建的数据流。

认证方式选择 GSSAPI 时，需要上传 keytab 与 Kerberos 文件：

如果 GSSAPI 连接失败，可以检查认证区域里的 DisablePAFXFAST 选项：开启后不使用 GSSAPI 的快速认证（PAFXFAST），部分 Kerberos 环境需要关闭快速认证才能连上。

2.1.3 映射

第三步配置样例数据，并映射已有表或新建表以导入 Kafka 数据。

2.1.3.1 基本配置

选择 CSV 源数据格式，并指定或配置样例数据。你需要指定或手动配置样例数据格式，以保证映射的准确。

在配置样例数据时，你可以从主题中选择也可以直接编辑数据。

如果 Kafka 主题（Topic）中有完整的数据，则从主题中选择它作为样例数据即可。
当主题中的数据不完整（如存在空值列）时，你可以选择直接编辑数据，按照目标表的字段结构给予我们一条样例数据。该样例数据只用于字段结构的解析与映射，不会被导入到目标表中。

注意！
使用图形化界面新创建的表，默认使用 MARS3 存储引擎。更多相关信息请见 MARS3

2.1.3.2 执行映射

简单地映射配置完成后，进行源数据字段与目标表字段的映射。此步骤已有表与新建表的操作有差异。

映射已有表

首先选择一条样例数据，再在映射关系中找到对应的目标字段，设置好之后点击保存，即可建立一条映射关系。

也可以直接一键自动映射，便捷的同时也保持映射的正确性。从 6.8.4 开始，自动映射生成的字段名会去掉多余的下划线，可读性更好。

自动映射完成后，映射关系会逐列显示输出值，可随时「清空」或「编辑」：
映射新建表

新建表除建立映射规则外，还需注意以下设置：当源字段值为 UNIX 时间戳且目标字段类型选择为 timestamptz 时，输出值右侧会显示复选框存储为 timestamp，勾选此框，输出值结果会自动转换为数据库的 timestamp 格式。

映射规则保存完之后，还需要确认「设备 ID（唯一标识）」与「时间列」两个设置：
- 设备 ID（唯一标识）：UPSERT 模式必须选择至少一个，用于识别需要更新的同一条数据；INSERT 模式可选，选择后用于数据分布和排序，不选时数据库默认使用第一列作为分布列。
- 时间列：分区表必须选择一个有效的时间列（目标表类型为 timestamptz）；非分区表不需要选择时间列。
新建表的映射页比已有表多出「设备 ID（唯一标识）」一列：

2.1.4 提交

最后，确认数据流所有的配置信息，有误返回修改，无误则提交。

2.1.4.1 已有表

页面展示了数据流所有的配置信息，确认信息后点击“提交”按钮。

2.1.4.2 新建表

比已有表更多地显示了唯一标识信息：分区表是「设备 ID + 时间列」的组合；非分区表不显示唯一标识信息。确认信息后点击“提交”按钮。

6.8.4 的提交页还会显示目标表的表类型与导入模式：

不论是已有表还是新建表，提交成功后都会进入到创建成功提示页面，到此 Kafka 数据流已成功创建。

2.2 JSON 数据流接入

这里主要描述与 CSV 格式接入的差异，集中在映射步骤。

首先，选择 JSON 数据格式并配置样例数据。
其次，JSON 列的索引使用 $["列名"] 的语义来索引，索引规则参照 JSONPath。
最后，JSON 数据支持多层级嵌套，YMatrix 支持选择不同层级的源数据字段创建映射规则。点击样例数据栏中 ⊕ 展开嵌套层级。

3 Kafka 数据流的管理操作

当你使用 YMatrix 图形化界面创建 Kafka 数据流后，数据会持续导入 YMatrix。如因服务调试等原因需要临时中断导入，可以在图形化界面中暂停或停止任务，之后恢复或启动。

对于运行中的 Kafka 数据流，可以从进程视图和任务视图两个入口管理导入：进程视图按 MatrixGate 进程管理其下的所有任务，任务视图按单个任务管理。

注意！
启停与暂停操作可能需要 1-10 秒生效，期间不要对该数据流进行其他操作。暂停或停止完成后，Kafka 数据停止导入；恢复或启动后数据流立即恢复导入，页面将在最多 5 秒后恢复显示最新的导入信息。

3.1 进程视图

进程视图按 MatrixGate 进程列出主机名、PID、类型、任务状态、吞吐和创建/退出时间；已退出的进程仍保留在列表中。

操作列的「查看」进入该进程的详情页，包含「性能」「活跃任务」「基本信息」三个标签：

性能：该进程的数据吞吐与执行耗时。
活跃任务：该进程下每个任务的数据源、目标表、状态与吞吐；每个任务的操作列可以「查看」任务日志、数据量和入库详情，并「暂停」「恢复」该任务的导入。
基本信息：该进程的配置信息。

列表行的「···」管理该进程下的所有任务，并可从列表中清除进程记录：

暂停所有任务：暂停该进程下的所有任务，不再接收导入。
恢复所有任务：恢复该进程下所有已暂停的任务，继续导入。
清除：从列表中清除该进程记录，不再显示。

3.2 任务视图

操作菜单下的扩展按钮可以进行启动、重设点位、停止和移除任务操作。

启动：启动已停止的任务，开始导入数据。
重设点位：对已停止的任务重设消费点位，详见「重设 Kafka 消费点位」一章。
停止：停止该任务，相关 MatrixGate 进程随之退出；之后可以再次启动。
移除任务：从列表中移除该任务，不再显示。

任务列表默认只显示「活跃于：最近 1 小时」的任务，需要查看更早的任务时切换时间筛选。

4 重设 Kafka 消费点位

如果需要从 Kafka 的另一个位置重新消费，例如误删数据后重新导入，或跳过积压的数据，可以对已停止的数据流重设消费点位。从 6.8.4 开始支持，可以在图形化界面的任务视图中操作，也可以使用 mxgate offset-reset 命令。

注意！
重设点位是离线操作。请先停止任务，并确认所有相关 MatrixGate 进程已退出，然后再重设点位。

注意！
重新消费遵循至少一次投递语义。请确保目标表能够处理重复数据，例如使用 UPSERT 或去重模式。

4.1 使用图形化界面

在任务视图中停止目标任务。
点击「重设点位」，在「重设消费点位」页面选择重新消费的位置：
- Kafka 保留的最早点位
- 最新点位（跳过积压）
- 指定时刻及之后的第一条消息
- 按分区指定点位，格式如 0:100,1:200
选择「指定时刻」时，需要指定日期和时间：
点击「生成计划」查看 Topic、消费组、目标表、保留范围起点、日志末端、已消费到、重置后从、待归档删除记录数等信息，确认无误后点击「确认重设」。
重设完成后，在任务视图中启动该任务，从新的点位继续消费。

4.2 使用命令行

mxgate offset-reset 对一条已停止的数据流重设消费点位。命令从 -C 指定的配置文件中读取数据流配置，只用它来定位数据流，不会修改配置文件。

前置条件

运行该数据流的所有 MatrixGate 进程都已退出，并且在命令完成前不会被 systemd、Kubernetes 等重新拉起。
使用 consumer group 时，命令会检查该 group 是否有活跃成员；simple consumer 仍在运行时命令检测不到，需要你自行确认进程已经退出。

指定目标位置（四种方式必须且只能指定一种）

参数	说明
`--to-earliest`	重设到 Kafka 仍保留的最早点位
`--to-latest`	重设到日志末端，跳过积压的数据
`--to-datetime <时间>`	重设到该时刻及之后的第一条消息，时间使用带时区的 RFC3339 格式，如 `2026-08-28T10:00:00+08:00`
`--to-offsets <分区:点位,...>`	按分区指定点位，如 `0:100,1:200`

选择要重设的数据流

--topic、--consumer-group、--job 只用于匹配配置文件中已有的数据流，不会覆盖或临时拼装配置。

配置文件中只有一条 Kafka 数据流时，命令自动选中它，不需要给出选择参数。
多条数据流时，用 --topic、--consumer-group、--job 精确指定其中一条。simple consumer 没有 consumer group，省略 --consumer-group。
只给出部分选择条件、没有匹配项或匹配到多条时，命令拒绝执行并列出候选配置。

其他参数

参数	说明
`-C, --config`	数据流的配置文件
`--dry-run`	只展示计划，不修改任何数据
`-y, --yes`	跳过交互确认，适合自动化场景
`--output`	输出格式：`text` 或 `json`。`json` 只向 stdout 输出一个对象，需配合 `--dry-run` 或 `-y` 使用

示例：预览计划

配置文件中只有一条数据流时，不需要给出选择参数：

$ mxgate offset-reset -C mxgate.conf --to-earliest --dry-run

命令会先提示这是离线操作，然后展示计划。以下输出为示例，实际输出与现场环境相关：

offset-reset is an offline command. Before continuing, make sure that
  - every mxgate instance running this flow is stopped and its process has exited;
  - nothing restarts it (systemd, Kubernetes, ...) until this command has succeeded or you have recovered from its error.
Only the consumer group is checked for live members; a simple consumer that is still running cannot be detected.

Offset reset plan of topic csv_test (job job_csv_to_public.demo_csv -> "public"."demo_csv", consumer group "", mode earliest):

partition  kafka-oldest   kafka-newest   consumed-to      consume-from   purge-rows
0          0              4              3                0              1

dry-run: nothing has been changed

计划中每行对应一个分区：kafka-oldest 与 kafka-newest 是 Kafka 保留的点位范围，consumed-to 是当前已消费到的位置，consume-from 是重设后开始消费的位置，purge-rows 是本次重设将从消费进度中移除的记录数。

示例：执行重设

配置文件中有多条数据流时，用选择参数指定一条，-y 跳过交互确认：

$ mxgate offset-reset -C mxgate.toml \
    --topic orders \
    --consumer-group mxgate-orders \
    --job j_orders \
    --to-offsets '0:100' \
    -y

执行与恢复

真实执行也会先展示计划，然后等待交互确认；确认后才会修改数据。
执行完成后，重新启动该数据流，从重设后的位置继续消费。
消费进度写入数据库与提交 Kafka consumer group 位点是两个独立操作，无法在同一个事务中完成。命令会在 30 秒内提交并读回验证 consumer group 位点；仍未确认时返回“部分成功”错误，并给出确切的 --to-offsets 恢复参数。此时按提示完成恢复，在确认 Kafka 位点已对齐前不要启动该数据流。

影响范围

不修改 Kafka 主题中的消息，也不改变 Kafka 的数据保留策略。
重设涉及的数据库修改在单个事务中整体提交或回滚。
未执行过重设的数据流，启动、消费和恢复语义不变。
被移除的消费进度和关联的错误记录会归档保留，可通过 reset_id 关联查询。

← 上一篇

MatrixGate 写入

Kafka SSL 加密连接

简体中文 English Русский