Kafka 连接过滤器 jsonpath 问题

Question

我正在使用 kafka connect 将数据从 kafka 接收到 mongo db。在此过程中，我尝试使用过滤器 (https://docs.confluent.io/current/connect/transforms/filter-confluent.html) 过滤一些记录。对于这个过滤器，我们需要给出一个谓词指定 JSON 路径作为条件来过滤来自 Kafka 的数据。

下面是我从 kafka 收到的 json 数据

**Input json:**

{
    "EventMetadata": {
            "ColumnsUpdated": "FirstName,LastName,Age"
        },
    "TotalBooking__c": 1
}

我有一个要求，它应该满足以下条件，如果满足以下条件，kafka connect 应该允许将此类记录插入 mongo db.

条件： 如果 TotalBooking__c > 0 并且 EventMetadata.ColumnsUpdated 包含任何“FirstName”或“LastName”值，它应该允许将此类记录插入 mongo db.

我尝试使用下面的 Kafka connect Filter operators

[?(@.TotalBooking__c > 0 && @.EventMetadata.ColumnsUpdated =~ /(?=\b\b)\b(FirstName|LastName)\b/)]

但是，只有当输入 EventMetadata.ColumnsUpdated 值只有“FirstName”或“LastName”时才有效。如果该值由逗号分隔（如上述输入 json 中给出的“FirstName,LastName,Age”），则它不起作用。

请帮我解决这个问题。

Answer 1

链接的 Confluent 文档显示 JayWay 的 JsonPath 实现在这里起作用，这一点很高兴知道，因为在功能和过滤器语法方面或多或少存在一些细微差别。

根据您的要求，我认为您使用正则表达式模式的方向正确。
但是，该模式需要匹配整个字符串（值），因为 =~ 执行正则表达式 match 而不是正则表达式 find。换句话说，我们需要允许在开头和结尾使用 .* 的可选文本（如果需要，可以使用更具体的模式）：

$[?(@.TotalBooking__c >0 && @.EventMetadata.ColumnsUpdated =~ /.*\b(FirstName|LastName)\b.*/)]

Kafka 连接过滤器 jsonpath 问题

Issue with Kafka connect filters jsonpath

jsonpath

apache-kafka