接收连接器后处理程序

创建自定义文档 ID 策略

如果内置文档 ID 加法器策略未涵盖您的使用案例，则可按照以下步骤定义自定义文档 ID 策略：

1. 创建一个实现接口 IdStrategy 并包含自定义配置逻辑的 Java 类。

2. 将此类编译为 JAR 文件。

3. 将已编译的 JAR 添加到所有Kafka Worker 的类路径/插件路径中。有关插件路径的更多信息，请参阅 Confluence 文档。

4. 将 document.id.strategy 设置更新为所有 Kafka Worker 中自定义类的完整类名。

有关 IdStrategy 接口的示例实现，请参阅包含用此连接器打包的 ID 策略实现的源代码目录。

允许列表通配符示例

您可以使用 * 通配符来匹配多个字段名称。以下示例配置将匹配如下字段：

• 名为 "city" 的顶级字段

• 名为“average”的字段，即任何以名称“wind_speed”开头的顶级字段的子文档。

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=city,wind_speed*.average

后处理器应用支持列表投影后，将输出以下记录：

{
  "city": "Springfield",
  "wind_speed_10m": {
    "average": 3,
  },
  "wind_speed_80m": {
    "average": 8,
  }
}

您可以使用 ** 通配符，匹配从您指定通配符开始的任何级别的对象。以下通配符匹配示例将投影任何包含名为“low”的字段的文档。

post.processor.chain=com.mongodb.kafka.connect.sink.processor.AllowListValueProjector
value.projection.type=AllowList
value.projection.list=**.low

应用投影的后处理器输出以下记录：

{
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "soil_conditions": {
    "temperature": {
      "high": 22,
      "low": 17,
      "units": "C"
    }
  }
}

阻止列表通配符示例

您可以使用通配符模式来匹配特定文档级别的字段，如以下区块列表配置示例所示：

post.processor.chain=com.mongodb.kafka.connect.sink.processor.BlockListValueProjector
value.projection.type=BlockList
value.projection.list=*.*.temperature

{
  "city": "Springfield",
  "temperature": {
    "high": 28,
    "low": 24,
    "units": "C"
  },
  "wind_speed_10m": {
    "average": 3,
    "units": "km/h"
  },
  "wind_speed_80m": {
    "average": 8,
    "units": "km/h"
  },
  "soil_conditions": {
    "moisture": {
      "average": 340,
      "units": "mm"
    }
  }
}

按映射重命名示例

RenameByMapping 帖子处理器设置指定一个或多个 JSON 对象，这些对象将与字符串匹配的字段分配给一个新名称。每个对象都包含 oldName 元素中要匹配的文本以及 newName 元素中的替换文本，如下表所述。

以下示例属性与关键文档的“位置”字段相匹配，并将其重命名为“国家/地区”：

field.renamer.mapping=[{"oldName":"key.location", "newName":"country"}]

此设置指示 RenameByMapping 后处理器将原始密钥文档转换为以下文档：

{
  "country": "Provence",
  "date_month": "October",
  "date_day": 17
}

您可以通过在 oldName 字段中指定具有附加字段名称的值文档，对值文档执行类似的字段名称分配，如下所示：

field.renamer.mapping=[{"oldName":"value.flapjacks", "newName":"crepes"}]

此设置指示 RenameByMapping 后处理器将原始值文档转换为以下文档：

{
  "crepes": {
    "purchased": 598,
    "size": "large"
  }
}

您还可以使用字符串格式的 JSON 数组在 field.renamer.mapping 属性中指定一个或多个映射，如以下设置所示：

field.renamer.mapping=[{ "oldName":"key.location", "newName":"city" }, { "oldName":"value.crepes", "newName":"flapjacks" }]

按正则表达式重命名

RenameByRegex 后处理器设置指定其应匹配的字段名和文本模式，以及匹配文本的替换值。您可以在包含下表所述字段的 JSON 对象中指定一个或多个重命名表达式：

以下示例设置指示后处理器执行以下操作：

• 匹配密钥文档中任何以“date”开头的字段名。在匹配字段集中，用 - 字符替换所有与模式 _ 匹配的文本。

• 匹配值文档中任何属于 crepes 的子文档的字段名。在匹配字段集中，用 quantity 替换所有与模式 purchased 匹配的文本。

field.renamer.regexp=[{"regexp":"^key\\.date.*$","pattern":"_","replace":"-"}, {"regexp":"^value\\.crepes\\..*","pattern":"purchased","replace":"quantity"}]

当连接器将后处理器应用于示例密钥文档和示例值文档时，它会输出以下内容：

关键文档

{
  "location": "Provence",
  "date-month": "October",
  "date-day": 17
}

值文档

{
  "crepes": {
    "quantity": 598,
    "size": "large"
  }
}