<a id="flatten-confluent"></a>

# 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:

```json
"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 `.`.

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

Before:

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

After:

```none
{
  "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.

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

Before:

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

After:

```none
{
  "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.

## 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](filter-ak.md#ak-filter), predicates
can conditionally filter out specific records. For details and examples, see [Predicates](filter-ak.md#predicates).