注意：此功能需要 Manticore Buddy。如果无法使用，请确保已安装 Buddy。

Manticore 支持通过 Kafka 源和物化视图与 Apache Kafka 实时数据摄取集成，实现实时数据索引和搜索。目前，apache/kafka 版本 3.7.0-4.1.0 已测试并支持。

开始之前，您需要：

1. 定义源： 指定 Manticore Search 将读取消息的 Kafka 主题。此设置包括代理的主机、端口和主题名称等详细信息。
2. 设置目标表： 选择一个 Manticore 实时表来存储传入的 Kafka 数据。
3. 创建物化视图： 设置物化视图（mv）以处理从 Kafka 到 Manticore Search 目标表的数据转换和映射。在这里，您将定义字段映射、数据转换以及传入数据流的任何过滤器或条件。

CREATE SOURCE <source name> [(column type, ...)] [source_options]

allowed_field_name 'original JSON key name with special symbols' type

price_field '$price' float    -- maps JSON key '$price' to field 'price_field'
field_123 '123field' text     -- maps JSON key '123field' to field 'field_123'

CREATE SOURCE kafka
(id bigint, term text, abbrev '$abbrev' text, GlossDef json)
type='kafka'
broker_list='kafka:9092'
topic_list='my-data'
consumer_group='manticore'
num_consumers='2'
batch=50

Query OK, 2 rows affected (0.02 sec)

CREATE TABLE destination_kafka
(id bigint, name text, short_name text, received_at text, size multi);

Query OK, 0 rows affected (0.02 sec)

CREATE MATERIALIZED VIEW <materialized view name>
TO <destination table name> AS
SELECT [column|function [as <new name>], ...] FROM <source name>

CREATE MATERIALIZED VIEW view_table
TO destination_kafka AS
SELECT id, term as name, abbrev as short_name,
       UTC_TIMESTAMP() as received_at, GlossDef.size as size FROM kafka

Query OK, 2 rows affected (0.02 sec)

数据以批次方式从 Kafka 传输到 Manticore Search，每次运行后批次被清空。对于跨批次的计算（如 AVG），请谨慎使用，因为批次处理可能导致结果不符合预期。

以下是基于上述示例的映射表：

SHOW SOURCES

+-------+
| name  |
+-------+
| kafka |
+-------+

SHOW SOURCE kafka;

+--------+-------------------------------------------------------------------+
| Source | Create Table                                                      |
+--------+-------------------------------------------------------------------+
| kafka  | CREATE SOURCE kafka                                               |
|        | (id bigint, term text, abbrev '$abbrev' text, GlossDef json)      |
|        | type='kafka'                                                      |
|        | broker_list='kafka:9092'                                          |
|        | topic_list='my-data'                                              |
|        | consumer_group='manticore'                                        |
|        | num_consumers='2'                                                 |
|        | batch=50                                                          |
+--------+-------------------------------------------------------------------+

SHOW MVS

+------------+
| name       |
+------------+
| view_table |
+------------+

SHOW MV view_table

+------------+--------------------------------------------------------------------------------------------------------+-----------+
| View       | Create Table                                                                                           | suspended |
+------------+--------------------------------------------------------------------------------------------------------+-----------+
| view_table | CREATE MATERIALIZED VIEW view_table TO destination_kafka AS                                            | 0         |
|            | SELECT id, term as name, abbrev as short_name, UTC_TIMESTAMP() as received_at, GlossDef.size as size   |           |
|            | FROM kafka                                                                                             |           |
+------------+--------------------------------------------------------------------------------------------------------+-----------+

ALTER MATERIALIZED VIEW view_table suspended=1

Query OK (0.02 sec)

您还可以为每个 Kafka 主题指定 partition_list。 这种方法的主要优点之一是能够通过 Kafka 实现表的 sharding（分片）。 为此，您应为每个分片创建一条独立的链：source → materialized view → destination table：

源：

CREATE SOURCE kafka_p1 (id bigint, term text)
  type='kafka' broker_list='kafka:9092' topic_list='my-data'
  consumer_group='manticore' num_consumers='1' partition_list='0' batch=50;
CREATE SOURCE kafka_p2 (id bigint, term text)
  type='kafka' broker_list='kafka:9092' topic_list='my-data'
  consumer_group='manticore' num_consumers='1' partition_list='1' batch=50;

目标表：

CREATE TABLE destination_shard_1 (id bigint, name text);
CREATE TABLE destination_shard_2 (id bigint, name text);

物化视图：

CREATE MATERIALIZED VIEW mv_1 TO destination_shard_1 AS SELECT id, term AS name FROM kafka_p1;
CREATE MATERIALIZED VIEW mv_2 TO destination_shard_2 AS SELECT id, term AS name FROM kafka_p2;

• 在此设置中，重新平衡必须手动管理。
• Kafka 默认不使用轮询（round-robin）策略分发消息。
• 若要在发送数据时实现类似轮询的分发，请确保您的 Kafka 生产者配置了：
  • parse.key=true
  • key.separator={your_delimiter}

否则，Kafka 会根据其内部规则分发消息，可能导致分区不均。

Kafka 在每个批次后或处理超时后提交偏移量。如果物化视图查询过程中进程意外停止，可能会出现重复条目。为避免此情况，请在您的模式中包含 id 字段，使 Manticore Search 能防止表中出现重复。

• 工作线程初始化： 配置源和物化视图后，Manticore Search 会设置专用工作线程处理来自 Kafka 的数据摄取。
• 消息映射： 根据源配置的模式映射消息，将其转换为结构化格式。
• 批处理： 将消息分组为批次以提高处理效率。批次大小可根据性能和延迟需求调整。
• 缓冲： 映射后的数据批次存储在缓冲表中，以便高效批量操作。
• 物化视图处理： 对缓冲表中的数据应用视图逻辑，执行任何转换或过滤。
• 数据传输： 处理后的数据随后传输到目标实时表。
• 清理： 每个批次处理后清空缓冲表，确保为下一批数据做好准备。

注意：与 DBeaver 的集成需要 Manticore Buddy。如果无法正常工作，请确保已安装 Buddy。

DBeaver 是一款 SQL 客户端软件应用程序和数据库管理工具。对于 MySQL 数据库，它通过 JDBC 驱动程序应用 JDBC 应用程序编程接口进行交互。

Manticore 允许您使用 DBeaver 以与操作 MySQL 数据库相同的方式处理存储在 Manticore 表中的数据。目前，推荐使用并测试过的版本是 25.2.0。其他版本可能可用，但可能会引入问题。

要开始在 DBeaver 中使用 Manticore，请按照以下步骤操作：

• 在 DBeaver 的界面中选择 New database connection 选项
• 选择 SQL -> MySQL 作为 DBeaver 的数据库驱动程序
• 设置 Server host 和 Port 选项，对应您的 Manticore 实例的主机和端口（保持 database 字段为空）
• 设置认证凭据为 root/<empty password>

由于 Manticore 并不完全支持 MySQL，因此在使用 Manticore 时，DBeaver 的部分功能可用。

您将能够：

• 查看、创建、删除和重命名表
• 添加和删除表列
• 插入、删除和更新列数据

您将无法：

• 使用数据库完整性检查机制（MyISAM 将被设置为唯一可用的存储引擎）
• 使用 MySQL 存储过程、触发器、事件等
• 管理数据库用户
• 设置其他数据库管理选项

某些 MySQL 数据类型当前不被 Manticore 支持，因此在使用 DBeaver 创建新表时无法使用。此外，一些支持的数据类型会被转换为最相似的 Manticore 类型，转换时会忽略类型精度。以下是支持的 MySQL 数据类型及其映射到的 Manticore 类型列表：

• BIGINT UNSIGNED => bigint
• BOOL => boolean
• DATE、DATETIME、TIMESTAMP => timestamp
• FLOAT => float
• INT => int
• INT UNSIGNED、SMALLINT UNSIGNED、TINYINT UNSIGNED、BIT => uint
• JSON => json
• TEXT、LONGTEXT、MEDIUMTEXT、TINYTEXT、BLOB、LONGBLOB、MEDIUMBLOB、TINYBLOB => text
• VARCHAR、LONG VARCHAR、BINARY、CHAR、VARBINARY、LONG VARBINARY => string

您可以在这里找到有关 Manticore 数据类型的更多详细信息。

Manticore 能够处理 DATE、DATETIME 和 TIMESTAMP 数据类型，但这需要启用 Manticore 的 Buddy。否则，尝试操作这些类型将导致错误。

请注意，TIME 类型不被支持。

• DBeaver 的 Preferences -> Connections -> Client identification 选项不得关闭或被覆盖。 为了正确与 DBeaver 配合工作，Manticore 需要区分其请求与其他请求。为此，它使用 DBeaver 在请求头中发送的客户端通知信息。禁用客户端通知将破坏该检测，从而影响 Manticore 的正常功能。

• 当您第一次尝试更新表中的数据时，会看到 No unique key 弹出消息，并被要求定义自定义唯一键。 当您看到此消息时，请执行以下步骤：

  • 选择 Custom Unique Key 选项
  • 在列列表中仅选择 id 列
  • 点击 Ok

之后，您将能够安全地更新数据。

注意：与 Apache Superset 的集成需要 Manticore Buddy。如果无法正常工作，请确保已安装 Buddy。

Apache Superset 是一个现代的数据探索和可视化平台。Manticore Search 可以与 Apache Superset 集成，允许用户充分利用 Manticore Search 的强大功能进行数据分析和可视化。

注意：此集成仅适用于 Apache Superset 3.x 分支，尚不支持版本 4 或更高版本。

在将 Manticore Search 与 Apache Superset 集成之前，请确保：

1. Manticore Search（版本 6.2.0 或更高）已正确安装并配置在您的服务器上。请参阅官方 Manticore Search 安装指南获取帮助。
2. Apache Superset 已在您的系统上设置。请按照官方 Apache Superset 安装指南进行安装。

要将 Manticore Search 连接到 Apache Superset：

1. 登录到您的 Apache Superset 仪表板。
2. 点击顶部导航栏中的“+”，选择“数据库连接”。
3. 点击“+ DATABASE”添加新数据库。
4. 选择“MySQL”作为数据库类型。
5. 在配置页面中，提供以下详细信息：
   • 数据库名称：数据库的名称（例如，“Manticore”）
   • 主机：您的 Manticore Search 服务器主机
   • 端口：Manticore Search 运行的端口
   • 用户名：您的 Manticore Search 用户名
   • 显示名称：连接的显示名称
6. 点击“Connect”验证连接并保存。

将 Manticore Search 连接到 Apache Superset 后，您可以使用 Manticore 数据创建图表和仪表板：

1. 在 Apache Superset 仪表板中，点击顶部导航栏的“Dashboards”，选择“New dashboard”。
2. 打开仪表板，点击“+”图标添加新图表。
3. 选择连接到 Manticore Search 的数据集。
4. 选择您想创建的图表类型（例如，条形图、折线图、饼图）。
5. 使用 Apache Superset 的查询构建器或编写 SQL 查询，从您的 Manticore Search 数据库中获取数据。
6. 根据需要自定义图表的外观、标签和其他设置。
7. 点击“UPDATE CHART”更新图表。
8. 为图表添加名称，点击“Save”将其永久保存到仪表板中。

通过 Apache Superset 使用 Manticore Search 时，您可以：

• 查看和查询 Manticore Search 表中的数据
• 基于 Manticore 数据创建各种类型的可视化
• 使用这些可视化构建和自定义仪表板

在通过 Apache Superset 使用时，Manticore Search 的数据类型会映射到 MySQL 数据类型。有关支持的数据类型及其映射的更多详细信息，请参阅 Manticore Search 文档。

• 通过 Apache Superset 使用 Manticore Search 时，某些高级 MySQL 功能可能不可用。
• 与直接使用 MySQL 数据库相比，某些数据操作可能受到限制。

通过遵循这些步骤和指南，您可以有效地将 Manticore Search 与 Apache Superset 集成，实现强大的数据探索和可视化功能。

有关 Apache Superset 和 Manticore Search 的更多信息，请参阅以下资源：

• Apache Superset 集成