Kafka Connect Transform Reference

This document is the configuration reference guide for Kafka SMTs developed by OpenG2P, that can be used on OpenSearch Sink Connectors.

Following is a list of some of the other transformations available on the OpenSearch Connectors, apart from the ones developed by OpenG2P:

• Apache Kafka Connect SMTs.

• Debezium Kafka Connect Transformations.

Transformations

DynamicNewField

Class name:

• org.openg2p.reporting.kafka.connect.DynamicNewField$Key - Applies transform only to the Key of Kafka Connect Record.

• org.openg2p.reporting.kafka.connect.DynamicNewField$Value - Applies transform only to the Value of Kafka Connect Record.

Description:

• This transformation can be used to query external data sources to retrieve new fields and add them to the current record, based on the values of some existing fields of this record.

• Currently, only Elasticsearch-based queries are supported. This means any index on Elasticsearch(or OpenSearch) can be queried and some new fields can be populated based on fields from the current record.

  • Some selected from the current record will be taken. ES will be queried for records where the selected field values match. The top response will be picked. Fields from that response can be added back to the current record.

Configuration:

Field name	Field title	Description	Default Value
query.type	Query Type	This is the type of query made to retrieve new field values. Supported values: es (Elasticsearch based).	es
input.fields	Input Fields	List of comma-separated fields that will be considered as input fields in the current record. Nested input fields are supported, like: (where profile is json that contains name and birthdate fields) profile.name,profile.birthdate
output.fields	Output Fields	List of comma-separated fields to be added to this record.
input.default.values	Input Default Values	List of comma-separated values to give in place of the input fields when an input field is empty or null. Length of this has to match that of input.fields.
es.index	ES Index	Elasticsearch(or OpenSearch) index to query for.
es.input.fields	ES Input Fields	List of comma-separated fields, to be queried on the ES index, each of which maps to the fields on input.fields. Length of this has to match that of input.fields.
es.output.fields	ES Output Fields	List of comma-separated fields, to be retrieved from the ES query response document, each of which maps to the fields on output.fields. Length of this has to match that of output.fields.
es.input.query.add.keyword	ES Input Query Add Keyword	Whether or not to add .keyword to the es.input.fields during the term query. Supported values: true / false .	false
es.security.enabled	ES Security Enabled	If this value is given as true, then Security is enabled on ES.
es.url	ES Url	Elasticsearch/OpenSearch base URL.
es.username	ES Username
es.password	ES Password

DynamicNewFieldInsertBack

Class name:

• org.openg2p.reporting.kafka.connect.DynamicNewFieldInsertBack$Key - Applies transform only to the Key of Kafka Connect Record.

• org.openg2p.reporting.kafka.connect.DynamicNewFieldInsertBack$Value - Applies transform only to the Value of Kafka Connect Record.

Description:

• This transformation can be used to add additional data to documents of different index.

• If record matches the configured condition, the given data will be updated into the record with given id.

Configuration:

Field name	Field title	Description	Default Value
query.type	Query Type	This is the type of query made to retrieve new field values. Supported values: es (Elasticsearch based).	es
id.expr	ID Jq Expression	Jq expression to evaluate the ID of the external document into which the data is supposed to be updated.
condition	Condition	Jq expression that evaluates to a boolean value which decides whether or not to update.
value	Value	Jq expression of the value, that evaluates to a JSON, that is to be updated into the external document.
es.index	ES Index	Elasticsearch(or OpenSearch) index to update into.
es.security.enabled	ES Security Enabled	If this value is given as true, then Security is enabled on ES.
es.url	ES Url	Elasticsearch/OpenSearch base URL.
es.username	ES Username
es.password	ES Password

ApplyJq

Class name:

• org.openg2p.reporting.kafka.connect.ApplyJq$Key - Applies transform only to the Key of Kafka Connect Record.

• org.openg2p.reporting.kafka.connect.ApplyJq$Value - Applies transform only to the Value of Kafka Connect Record.

Description:

• This transformation applies the given Jq expression on the current record and replace the current record with the result from Jq.

• This transformation can be used for operations like extracting, merging, removing, and/or renaming fields.

• For example:

  • "expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}", : The expr field should contain a valid Jq expression.

Configuration:

Field name	Field title	Description	Default value
expr	Expression	Jq expression to be applied.
behavior.on.error	Behaviour on error	What to do when encountering error applying Jq expression. Possible values: halt : Throws exception upon encountering error. ignore : Ignores any errors encountered.	halt

StringToJson

Class name:

• org.openg2p.reporting.kafka.connect.StringToJson$Key - Applies transform only to the Key of Kafka Connect Record.

• org.openg2p.reporting.kafka.connect.StringToJson$Value - Applies transform only to the Value of Kafka Connect Record.

Description:

• This transformation can be used to convert JSON string, present in a field in the record, to JSON. Example:

  {"profile": "{\"name\":\"Temp\"}"} -> {"profile": {"name": "Temp"}}

• Currently, this transform only works in schemaless mode. (value.converter.schemas.enable=false).

Configuration

Field name	Field title	Description	Default Value
input.field	Input Field	Input Field that contains JSON string.

TimestampConverterAdv

Class name:

• org.openg2p.reporting.kafka.connect.TimestampConverterAdv$Key - Applies transform only to the Key of Kafka Connect Record.

• org.openg2p.reporting.kafka.connect.TimestampConverterAdv$Value - Applies transform only to the Value of Kafka Connect Record.

Description:

• This transformation can be used to convert a Timestamp present in a field in the record, to another format. Example:

  {"create_date": 1723667415} -> {"profile": "2024-08-14'T'20:30:50.069'Z'"}

• Currently, the output can only be in the form of a string.

Configuration

Field name	Field title	Description	Default Value
field	Input Field	Input Field that contains the Timestamp.
input.type	Input Type	Supported values: milli_sec (Input is present as milliseconds since epoch) micro_sec (Input is present as microseconds since epoch. Useful for converting Datetime field of PostgreSQL) days_epoch (Input is present as days since epoch. Useful for converting Date field of PostgreSQL)	milli_sec
output.type	Output Type	Supported values: string (Gives output as string)	string
output.format	Output Format	Format of string output	yyyy-MM-dd'T'HH:mm:ss.SSS'Z'

TimestampSelector

Class name:

• org.openg2p.reporting.kafka.connect.TimestampSelector$Key - Applies transform only to the Key of Kafka Connect Record.

• org.openg2p.reporting.kafka.connect.TimestampSelector$Value - Applies transform only to the Value of Kafka Connect Record.

Description:

• This transformation can be used to create a new timestamp field, whose value can be selected from other fields, in the order of whichever is not empty first. Example: (when ts.order is profile.write_date,profile.create_date)

  {"profile": {"create_date": 7415, "write_date": null}} -> {"@timestamp_gen": 7415, "profile": {"create_date": 7415, "write_date": null}}
  {"profile": {"create_date": 2945, "write_date": 3442}} -> {"@timestamp_gen": 3442, "profile": {"create_date": 2945, "write_date": 3442}}

Configuration

Field name	Field title	Description	Default Value
ts.order	Timestamp order	List of comma-separated fields to select output from. The output will be selected based on whichever field in the order is not null first. Nested fields are supported.
output.field	Output Field	Name of the output field into which the selected timestamp is put.	@ts_generated

TriggerDeduplication

Class name:

• org.openg2p.reporting.kafka.connect.TriggerDeduplication$Key - Applies transform only to the Key of Kafka Connect Record.

• org.openg2p.reporting.kafka.connect.TriggerDeduplication$Value - Applies transform only to the Value of Kafka Connect Record.

Description:

• This transformation can be used to trigger deduplication when there is a change in any one of the configured fields.

• This transformation is best used before applying any other transformation.

Configuration

Field name	Field title	Description	Default Value
deduplication.base.url	Base URL of Deduplicator Service
dedupe.config.name	Dedupe Config name	Name of config used for deduplication by deduplicator	default
id.expr	ID Jq Expression	Jq expression that evaluates the ID of the document that is to be deduplicated	.payload.after.id
before.expr	Before Jq Expression	Jq expression that evaluates the before part of the change. (Used to compare fields with the after part of the change).	.payload.before
after.expr	After Jq Expression	Jq expression that evaluates the after part of the change. (Used to compare fields with the before part of the change).	.payload.after
wait.before.exec.secs	Wait before Exec (in secs)	Time to wait (in secs) before starting deduplication. Useful so that the transformations get applied and the record get indexed into OpenSearch	10

Source code

https://github.com/OpenG2P/openg2p-reporting/tree/develop/opensearch-kafka-connector