接收器连接器后处理器

创建自定义文档 ID 策略

如果内置的文档 ID 添加策略不适用于您的用例，您可以通过以下步骤定义自定义文档 ID 策略。

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

2. 将类编译成 JAR 文件。

3. 将编译好的 JAR 文件添加到所有 Kafka 工作进程的类路径或插件路径中。有关插件路径的更多信息，请参阅Confluent 文档。

4. 更新所有 Kafka 工作进程中的 document.id.strategy 设置为您的自定义类的全类名。

例如，关于 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 元素中的替换文本，如下表所述。

以下示例属性匹配键文档中的 "location" 字段并将其重命名为 "country"。

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子文档的任何字段。在匹配字段集中，将所有匹配模式purchased的文本替换为quantity。

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"
  }
}