Sink Connector Procesadores posteriores

Crear una estrategia personalizada de identificación de documentos

Si las estrategias de funcionalidad incorporada de adición de ID de documento no cubren tu caso de uso, puedes definir una estrategia de ID de documento personalizada siguiendo los pasos que se indican a continuación:

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

2. Compila la clase en un archivo JAR.

3. Agrega el JAR compilado a la ruta de clase / ruta del plugin para todos los trabajadores de Kafka. Para obtener 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 ejemplos de implementaciones de la interfaz IdStrategy, consulta el directorio de código fuente que contiene implementaciones de estrategia id empaquetadas con el conector.

Permitir ejemplos de comodines de lista

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 denominado "ciudad"

• Los campos llamadas "promedio" que son subdocumentos de cualquier campo de nivel superior que comience con el nombre "velocidad_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 aplique la proyección de la lista de permitidos, produce 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 procesador posterior 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 en la lista de bloqueados

Puede utilizar los patrones de comodines para hacer coincidir campos a un nivel de documento específico, como se muestra en el siguiente ejemplo de configuración de lista de bloqueos:

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 los campos que coinciden con una string a un nuevo nombre. Cada objeto contiene el texto para hacer 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 como "país":

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

Esta configuración da instrucciones al post-procesador RenameByMapping para convertir el documento de 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 indica al procesador posterior de RenameByMapping que transforme el documento de valor original al siguiente documento:

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

También puedes especificar una o más asignaciones en la propiedad field.renamer.mapping usando un arreglo JSON en formato de string 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 instruye al pos-procesador a realizar lo siguiente:

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

• Coincide con cualquier nombre de campo en el documento de valores que sean subdocumentos de crepes. En el conjunto de campos coincidentes, sustituya 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 posprocesador al documento clave de ejemplo y al documento de valor de ejemplo, produce lo siguiente:

Documento clave

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

Documento de valor

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