title_wip

[Edit this page](https://github.com/ClickHouse/ClickHouse/tree/master/docs/en/interfaces/formats/Avro/AvroConfluent.md)

|  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- |
| Input Output Alias|  |  |  | | --- | --- | --- | | ✔ ✗  | | | | | |

## Description[​](#description "Direct link to Description")

[Apache Avro](https://avro.apache.org/) is a row-oriented serialization format that uses binary encoding for efficient data processing. The `AvroConfluent` format supports decoding single-object, Avro-encoded Kafka messages serialized using the [Confluent Schema Registry](https://docs.confluent.io/current/schema-registry/index.html) (or API-compatible services).

Each Avro message embeds a schema ID that ClickHouse automatically resolves by querying the configured schema registry. Once resolved, schemas are cached for optimal performance.

## Data type mapping[​](#data-type-mapping "Direct link to Data type mapping")

The table below shows all data types supported by the Apache Avro format, and their corresponding ClickHouse [data types](/docs/sql-reference/data-types) in `INSERT` and `SELECT` queries.

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Avro data type `INSERT` ClickHouse data type Avro data type `SELECT`|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `boolean`, `int`, `long`, `float`, `double` [Int(8\16\32)](/docs/sql-reference/data-types/int-uint), [UInt(8\16\32)](/docs/sql-reference/data-types/int-uint) `int`| `boolean`, `int`, `long`, `float`, `double` [Int64](/docs/sql-reference/data-types/int-uint), [UInt64](/docs/sql-reference/data-types/int-uint) `long`| `boolean`, `int`, `long`, `float`, `double` [Float32](/docs/sql-reference/data-types/float) `float`| `boolean`, `int`, `long`, `float`, `double` [Float64](/docs/sql-reference/data-types/float) `double`| `bytes`, `string`, `fixed`, `enum` [String](/docs/sql-reference/data-types/string) `bytes` or `string` \*| `bytes`, `string`, `fixed` [FixedString(N)](/docs/sql-reference/data-types/fixedstring) `fixed(N)`| `enum` [Enum(8\16)](/docs/sql-reference/data-types/enum) `enum`| `array(T)` [Array(T)](/docs/sql-reference/data-types/array) `array(T)`| `map(V, K)` [Map(V, K)](/docs/sql-reference/data-types/map) `map(string, K)`| `union(null, T)`, `union(T, null)` [Nullable(T)](/docs/sql-reference/data-types/date) `union(null, T)`| `union(T1, T2, …)` \*\* [Variant(T1, T2, …)](/docs/sql-reference/data-types/variant) `union(T1, T2, …)` \*\*| `null` [Nullable(Nothing)](/docs/sql-reference/data-types/special-data-types/nothing) `null`| `int (date)` \*\*\* [Date](/docs/sql-reference/data-types/date), [Date32](/docs/sql-reference/data-types/date32) `int (date)` \*\*\*| `long (timestamp-millis)` \*\*\* [DateTime64(3)](/docs/sql-reference/data-types/datetime) `long (timestamp-millis)` \*\*\*| `long (timestamp-micros)` \*\*\* [DateTime64(6)](/docs/sql-reference/data-types/datetime) `long (timestamp-micros)` \*\*\*| `bytes (decimal)` \*\*\* [DateTime64(N)](/docs/sql-reference/data-types/datetime) `bytes (decimal)` \*\*\*| `int` [IPv4](/docs/sql-reference/data-types/ipv4) `int`| `fixed(16)` [IPv6](/docs/sql-reference/data-types/ipv6) `fixed(16)`| `bytes (decimal)` \*\*\* [Decimal(P, S)](/docs/sql-reference/data-types/decimal) `bytes (decimal)` \*\*\*| `string (uuid)` \*\*\* [UUID](/docs/sql-reference/data-types/uuid) `string (uuid)` \*\*\*| `fixed(16)` [Int128/UInt128](/docs/sql-reference/data-types/int-uint) `fixed(16)`| `fixed(32)` [Int256/UInt256](/docs/sql-reference/data-types/int-uint) `fixed(32)`| `record` [Tuple](/docs/sql-reference/data-types/tuple) `record` | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | |

\* `bytes` is default, controlled by setting [`output_format_avro_string_column_pattern`](/docs/operations/settings/formats#output_format_avro_string_column_pattern)

\*\* The [Variant type](/docs/sql-reference/data-types/variant) implicitly accepts `null` as a field value, so for example the Avro `union(T1, T2, null)` will be converted to `Variant(T1, T2)`.
As a result, when producing Avro from ClickHouse, we have to always include the `null` type to the Avro `union` type set as we don't know if any value is actually `null` during the schema inference.

\*\*\* [Avro logical types](https://avro.apache.org/docs/current/spec.html#Logical+Types)

Unsupported Avro logical data types:

- `time-millis`
- `time-micros`
- `duration`

## Format settings[​](#format-settings "Direct link to Format settings")

|  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Setting Description Default|  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `input_format_avro_allow_missing_fields` Whether to use a default value instead of throwing an error when a field is not found in the schema. `0`| `input_format_avro_null_as_default` Whether to use a default value instead of throwing an error when inserting a `null` value into a non-nullable column. `0`| `format_avro_schema_registry_url` The Confluent Schema Registry URL. For basic authentication, URL-encoded credentials can be included directly in the URL path.  | | | | | | | | | | | |

## Examples[​](#examples "Direct link to Examples")

### Using a schema registry[​](#using-a-schema-registry "Direct link to Using a schema registry")

To read an Avro-encoded Kafka topic using the [Kafka table engine](/docs/engines/table-engines/integrations/kafka), use the `format_avro_schema_registry_url` setting to provide the URL of the schema registry.

```
CREATE TABLE topic1_stream
(
    field1 String,
    field2 String
)
ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'kafka-broker',
kafka_topic_list = 'topic1',
kafka_group_name = 'group1',
kafka_format = 'AvroConfluent',
format_avro_schema_registry_url = 'http://schema-registry-url';

SELECT * FROM topic1_stream;
```

#### Using basic authentication[​](#using-basic-authentication "Direct link to Using basic authentication")

If your schema registry requires basic authentication (e.g., if you're using Confluent Cloud), you can provide URL-encoded credentials in the `format_avro_schema_registry_url` setting.

```
CREATE TABLE topic1_stream
(
    field1 String,
    field2 String
)
ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'kafka-broker',
kafka_topic_list = 'topic1',
kafka_group_name = 'group1',
kafka_format = 'AvroConfluent',
format_avro_schema_registry_url = 'https://<username>:<password>@schema-registry-url';
```

## Troubleshooting[​](#troubleshooting "Direct link to Troubleshooting")

To monitor ingestion progress and debug errors with the Kafka consumer, you can query the [`system.kafka_consumers` system table](/docs/operations/system-tables/kafka_consumers). If your deployment has multiple replicas (e.g., ClickHouse Cloud), you must use the [`clusterAllReplicas`](/docs/sql-reference/table-functions/cluster) table function.