Kafka Connect Flatten (Confluent) SMT Usage Reference for Confluent Cloud

This topic describes the Confluent Single Message Transformation (SMT) io.confluent.connect.transforms.Flatten.

Description

The Confluent Flatten SMT transforms a record’s key or value by converting complex nested structures (such as structs) into flat, top-level fields using a delimiter. This SMT also handles nested maps encountered within struct schemas. You can set the behaviour.on.nested.map property to either stop processing records (FAIL) if such nested maps are found, or to include them as is (INCLUDE) without flattening their contents. The default delimiter is the period . character.

Note

The behaviour.on.nested.map property applies only to nested maps found within a struct schema. For schemaless data, all nested maps are flattened recursively, regardless of this setting.

You can use the concrete transformation type designed for the record key (io.confluent.connect.transforms.Flatten$Key) or value (io.confluent.connect.transforms.Flatten$Value).

When using an Avro schema, you cannot use the default SMT delimiter period .. Use an underscore _ for the delimiter instead. Use the following properties to change the delimiter:

"transforms": "flatten",
"transforms.flatten.type": "io.confluent.connect.transforms.Flatten$Value",
"transforms.flatten.delimiter": "_"

Note

The Flatten SMT does not flatten arrays. Arrays and their contents are preserved as is.

JSON example

This configuration snippet shows how to use Flatten to flatten a record’s value while including a nested map without flattening its internal structure. This snippet uses the default delimiter ..

"transforms": "flatten",
"transforms.flatten.type": "io.confluent.connect.transforms.Flatten$Value",
"transforms.flatten.behaviour.on.nested.map": "INCLUDE",
"transforms.flatten.delimiter": "."

Before:

{
  "user": {
      "id": "1234",
      "attributes": { "role": "admin", "dept": "engineering" },
      "profile": {
          "address": {
          "city": "Bangalore",
          "metadata": { "zone": "east", "block": "A" }
         },
      "age": 30
      }
     }
 }

After:

{
  "user.id": "1234",
  "user.attributes": { "role": "admin", "dept": "engineering" },
  "user.profile.address.city": "Bangalore",
  "user.profile.address.metadata": { "zone": "east", "block": "A" },
  "user.profile.age": 30
 }

If transforms.flatten.behaviour.on.nested.map is set to FAIL, the transformation fails the connector task at runtime when it encounters a nested map within a struct schema.

Avro example

The Avro schema specification only allows alphanumeric characters and the underscore _ character in field names. The following configuration snippet shows how to use Flatten to concatenate field names with the underscore _ delimiter character.

"transforms": "flatten",
"transforms.flatten.type": "io.confluent.connect.transforms.Flatten$Value",
"transforms.flatten.behaviour.on.nested.map": "INCLUDE",
"transforms.flatten.delimiter": "_"

Before:

{
  "user": {
      "id": "1234",
      "attributes": { "role": "admin", "dept": "engineering" },
      "profile": {
          "address": {
          "city": "Bangalore",
          "metadata": { "zone": "east", "block": "A" }
         },
      "age": 30
      }
     }
 }

After:

{
  "user_id": "1234",
  "user_attributes": { "role": "admin", "dept": "engineering" },
  "user_profile_address_city": "Bangalore",
  "user_profile_address_metadata": { "zone": "east", "block": "A" },
  "user_profile_age": 30
 }

If transforms.flatten.behaviour.on.nested.map is set to FAIL, the transformation fails the connector task at runtime when it encounters a nested map within a struct schema.

Tip

For additional examples, see Flatten for managed connectors.

Properties

Name	Description	Type	Default	Valid Values	Importance
delimiter	Delimiter to insert between field names from the input record when generating field names for the output record.	string	.		medium
behaviour.on.nested.map	Behavior when a nested map is found within a struct schema. FAIL: Fail the transformation if a nested map is encountered in a struct schema. INCLUDE: Include the nested map as is without flattening the nested map. The key for the map must be a string.	string	FAIL	FAIL, INCLUDE	medium

Predicates

Transformations can be configured with predicates so that the transformation is applied only to records which satisfy a condition. You can use predicates in a transformation chain and, when combined with the Kafka Connect Filter (Kafka) SMT Usage Reference for Confluent Cloud, predicates can conditionally filter out specific records. For details and examples, see Predicates.