Skip to content

Latest commit

 

History

History
147 lines (89 loc) · 12.9 KB

kafka-connect-transform-reference.md

File metadata and controls

147 lines (89 loc) · 12.9 KB
Raw

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:

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 nameField titleDescriptionDefault Value
query.typeQuery Type

This is the type of query made to retrieve new field values.

Supported values:

  • es (Elasticsearch based).
es
input.fieldsInput 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.fieldsOutput FieldsList of comma-separated fields to be added to this record.
input.default.valuesInput Default ValuesList 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.indexES IndexElasticsearch(or OpenSearch) index to query for.
es.input.fieldsES Input FieldsList 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.fieldsES Output FieldsList 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.keywordES Input Query Add KeywordWhether or not to add .keyword to the es.input.fields during the term query. Supported values: true / false .false
es.security.enabledES Security EnabledIf this value is given as true, then Security is enabled on ES.
es.urlES UrlElasticsearch/OpenSearch base URL.
es.usernameES Username
es.passwordES 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 nameField titleDescriptionDefault Value
query.typeQuery Type

This is the type of query made to retrieve new field values.

Supported values:

  • es (Elasticsearch based).
es
id.exprID Jq ExpressionJq expression to evaluate the ID of the external document into which the data is supposed to be updated.
conditionConditionJq expression that evaluates to a boolean value which decides whether or not to update.
valueValueJq expression of the value, that evaluates to a JSON, that is to be updated into the external document.
es.indexES IndexElasticsearch(or OpenSearch) index to update into.
es.security.enabledES Security EnabledIf this value is given as true, then Security is enabled on ES.
es.urlES UrlElasticsearch/OpenSearch base URL.
es.usernameES Username
es.passwordES 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 nameField titleDescriptionDefault value
exprExpressionJq expression to be applied.
behavior.on.errorBehaviour 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 nameField titleDescriptionDefault Value
input.fieldInput FieldInput 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 nameField titleDescriptionDefault Value
fieldInput FieldInput Field that contains the Timestamp.
input.typeInput 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.typeOutput Type

Supported values:

  • string (Gives output as string)
string
output.formatOutput FormatFormat 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 nameField titleDescriptionDefault Value
ts.orderTimestamp orderList 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.fieldOutput FieldName 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 nameField titleDescriptionDefault Value
deduplication.base.urlBase URL of Deduplicator Service
dedupe.config.nameDedupe Config nameName of config used for deduplication by deduplicatordefault
id.exprID Jq ExpressionJq expression that evaluates the ID of the document that is to be deduplicated
.payload.after.id
before.exprBefore Jq ExpressionJq expression that evaluates the before part of the change. (Used to compare fields with the after part of the change).
.payload.before
after.exprAfter Jq ExpressionJq 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.secsWait 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 OpenSearch10

Source code

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