NOTE: 此功能需要 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 为您的表实现分片。 为此，您应该为每个分片创建一条独立的source → 物化视图 → 目标表链：

源:

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 默认情况下不使用轮询策略分发消息。
• 若要在发送数据时实现类似轮询的分发，请确保您的 Kafka 生产者配置了：
  • parse.key=true
  • key.separator={your_delimiter}

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

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

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

NOTE: 与DBeaver的集成需要Manticore Buddy。如果不起作用，请确保已安装Buddy。

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

Manticore允许您使用DBeaver与存储在Manticore表中的数据进行操作，就像这些数据存储在MySQL数据库中一样。目前，已测试并推荐使用版本25.2.0。其他版本可能也能工作，但可能会引入问题。

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

• 在DBeaver的UI中选择新建数据库连接选项
• 选择SQL -> MySQL作为DBeaver的数据库驱动
• 设置服务器主机和端口选项，对应于您的Manticore实例的主机和端口（保持数据库字段为空）
• 设置root/<空密码>作为身份验证凭据

由于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的首选项 -> 连接 -> 客户端标识选项必须不被关闭或覆盖。 为了正确使用DBeaver，Manticore需要能够区分其请求与其他请求。为此，它使用DBeaver在请求头中发送的客户端通知信息。禁用客户端通知将破坏这种检测，从而影响Manticore的正确功能。

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

  • 选择自定义唯一键选项
  • 在列列表中仅选择id列
  • 点击确定

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

注意：与 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 数据创建各种类型的可视化
• 使用这些可视化构建和定制仪表板

Manticore Search 的数据类型在与 Apache Superset 交互时会映射为 MySQL 数据类型。有关支持的数据类型及其映射的详细信息，请参阅 Manticore Search 文档。

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

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

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

• Apache Superset 集成