Postprocesadores de conector de sumidero

Crear una estrategia de identificación de documento personalizada

Si las estrategias de adición de identificación de documento integradas no cubren su caso de uso, puede definir una estrategia de identificación de documento personalizada siguiendo los pasos a continuación:

1. Cree una clase Java que implemente la interfaz IdStrategy y contenga su lógica de configuración personalizada.

2. Compila la clase en un archivo JAR.

3. Añade el JAR compilado a la ruta de clases o del plugin de todos tus trabajadores de Kafka. Para más información sobre las rutas de plugins, consulta la documentación de Confluent.

4. Actualice la configuración document.id.strategy con el nombre de clase completo de su clase personalizada en todos sus trabajadores de Kafka.

Para ver ejemplos de implementaciones de la IdStrategy interfaz, consulte el directorio del código fuente que contiene las implementaciones de la estrategia de identificación empaquetadas con el conector.

Ejemplos de comodines de listas de permitidos

Puedes utilizar el comodín * para hacer coincidir varios nombres de campo. La siguiente configuración de ejemplo coincide con los siguientes campos:

• El campo de nivel superior llamado "ciudad"

• Los campos denominados "promedio" que son subdocumentos de cualquier campo de nivel superior que comience con el nombre "velocidad_del_viento".

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

Después de que el postprocesador aplica la proyección de la lista de permitidos, genera el siguiente registro:

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

Puedes usar el comodín **, que coincide con objetos en cualquier nivel a partir del nivel en el que especifiques el comodín. El siguiente ejemplo de coincidencia comodín proyecta cualquier documento que contenga el campo llamado "low".

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

El postprocesador que aplica la proyección genera el siguiente registro:

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

Ejemplo de comodín de lista de bloqueo

Puede utilizar patrones comodín para hacer coincidir campos en un nivel de documento específico como se muestra en el siguiente ejemplo de configuración de lista de bloqueo:

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

Ejemplo de cambio de nombre por mapeo

La configuración del posprocesador RenameByMapping especifica uno o más objetos JSON que asignan un nuevo nombre a los campos que coinciden con una cadena. Cada objeto contiene el texto que debe coincidir en el elemento oldName y el texto de reemplazo en el elemento newName, como se describe en la tabla a continuación.

La siguiente propiedad de ejemplo coincide con el campo "ubicación" de un documento clave y lo renombra a "país":

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

Esta configuración le indica al RenameByMapping postprocesador que transforme el documento clave original en el siguiente documento:

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

Puede realizar una asignación de nombre de campo similar para documentos de valor especificando el documento de valor con el nombre de campo adjunto en el campo oldName de la siguiente manera:

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

Esta configuración le indica al RenameByMapping postprocesador que transforme el documento de valor original en el siguiente documento:

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

También puede especificar una o más asignaciones en la propiedad field.renamer.mapping utilizando una matriz JSON en formato de cadena como se muestra en la siguiente configuración:

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

Cambiar nombre por expresión regular

La configuración del posprocesador RenameByRegex especifica los nombres de campo y los patrones de texto que debe coincidir, así como los valores de reemplazo para el texto coincidente. Puede especificar una o más expresiones de cambio de nombre en objetos JSON que contengan los campos descritos en la siguiente tabla:

La siguiente configuración de ejemplo le indica al postprocesador que realice lo siguiente:

• Coincida con cualquier nombre de campo del documento clave que comience por "fecha". En el conjunto de campos coincidentes, reemplace todo el texto que coincida con el patrón _ por el carácter -.

• Coincide con cualquier nombre de campo en el documento de valor que sea subdocumento de crepes. En el conjunto de campos coincidentes, reemplaza todo el texto que coincida con el patrón purchased por quantity.

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

Cuando el conector aplica el postprocesador al documento clave de ejemplo y al documento de valor de ejemplo, genera lo siguiente:

Documento clave

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

Documento de valor

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