2.5. 将操作应用到 Kafka 连接中的数据
如果要在 Kamelet 和 Kafka 主题之间传递数据执行操作,请在 Kamelet Binding 中将 action Kamelets 用作中间步骤。
2.5.1. 将事件数据路由到不同的目的地主题
当您配置到 Kafka 实例的连接时,您可以选择性地将主题信息从事件数据转换,以便事件路由到不同的 Kafka 主题。使用以下转换操作 Kamelets 之一:
-
regex Router - 使用正则表达式和替换字符串修改消息主题。例如,如果要删除主题前缀,请添加前缀或删除主题名称的一部分。配置 Regex Router Action Kamelet(
regex-router-action
)。 -
timestamp - 根据原始主题和消息的时间戳修改消息的主题。例如,在使用需要根据时间戳写入不同表或索引的 sink 时。例如,当您要将事件从 Kafka 写入 Elasticsearch 时,每个事件都需要根据事件本身中的信息进入不同的索引。配置 Timestamp Router Action Kamelet(
timestamp-router-action
)。 -
Message TimeStamp - 根据原始主题值修改消息主题,以及来自消息值的 timestamp 字段。配置 Message Timestamp Router Action Kamelet(
message-timestamp-router-action
)。 -
predicate - 通过配置 Predicate Filter Action Kamelet(
predicate-filter-action
),根据给定的 JSON 路径表达式过滤事件。
先决条件
-
您已创建了 Kamelet Binding,其中接收器是一个
kafka-sink
Kamelet,如 Kamelet Binding 的 Kafka 主题所述。 - 您知道要添加到 Kamelet Binding 中的哪一种转换。
流程
要转换目标主题,请在 Kamelet Binding 中使用一个转换操作 Kamelets 中的一个中间步骤。
有关如何在 Kamelet 中添加操作 Kamelet 的详细信息,请参阅在 Kamelet Binding 中添加操作。