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

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

This topic describes the Apache Kafka® Single Message
Transformation (SMT) `org.apache.kafka.connect.transforms.Flatten`.

## Description

The Flatten SMT flattens a nested data structure, generating names for each
field by concatenating the field names at each level with a configurable
delimiter character. This SMT applies to a `Struct` when a schema is present,
or to a `Map` for schemaless data. The default delimiter is the period
`.` character.

Use the concrete transformation type designed for the record key
(`org.apache.kafka.connect.transforms.Flatten$Key`) or value
(`org.apache.kafka.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": "org.apache.kafka.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

The following configuration snippet shows how to use `Flatten` to concatenate
field names with the period `.` delimiter character.

```json
"transforms": "flatten",
"transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
"transforms.flatten.delimiter": "."
```

Before:

```none
{
  "content": {
    "id": 42,
    "name": {
      "first": "David",
      "middle": null,
      "last": "Wong"
    }
  }
}
```

After:

```none
{
  "content.id": 42,
  "content.name.first": "David",
  "content.name.middle": null,
  "content.name.last": "Wong"
}
```

## 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": "org.apache.kafka.connect.transforms.Flatten$Value",
"transforms.flatten.delimiter": "_"
```

Before:

```none
{
  "content": {
    "id": 42,
    "name": {
      "first": "David",
      "middle": null,
      "last": "Wong"
    }
  }
}
```

After:

```none
{
  "content_id": 42,
  "content_name_first": "David",
  "content_name_middle": null,
  "content_name_last": "Wong"
}
```

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

## 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).