0
# Apache Flink SQL Avro Format Library
1

2
## Overview
3

4
The Apache Flink SQL Avro format library provides SQL support for Apache Avro data format within Flink applications. This library is a bundling and shading module that packages Apache Avro dependencies with appropriate relocations to avoid classpath conflicts, enabling seamless Avro format processing in Flink SQL pipelines.
5

6
## Package Information
7

8
- **Package Name**: org.apache.flink:flink-sql-avro
9
- **Package Type**: maven
10
- **Language**: Java
11
- **Version**: 2.1.0
12
- **License**: Apache-2.0
13
- **Installation**: Add to `pom.xml`:
14

15
```xml
16
<dependency>
17
    <groupId>org.apache.flink</groupId>
18
    <artifactId>flink-sql-avro</artifactId>
19
    <version>2.1.0</version>
20
</dependency>
21
```
22

23
## Core Imports
24

25
```java
26
import org.apache.flink.formats.avro.AvroDeserializationSchema;
27
import org.apache.flink.formats.avro.AvroSerializationSchema;
28
import org.apache.flink.formats.avro.AvroRowDataDeserializationSchema;
29
import org.apache.flink.formats.avro.AvroRowDataSerializationSchema;
30
import org.apache.flink.formats.avro.AvroFormatOptions;
31
import org.apache.flink.formats.avro.AvroFormatOptions.AvroEncoding;
32
import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
33
import org.apache.flink.formats.avro.typeutils.AvroTypeInfo;
34
import org.apache.flink.formats.avro.typeutils.GenericRecordAvroTypeInfo;
35
```
36

37
## Basic Usage
38

39
### SQL Table with Avro Format
40

41
```java { .api }
42
// Create table with Avro format in Flink SQL
43
String createTableSQL = """
44
    CREATE TABLE avro_table (
45
        id INT,
46
        name STRING,
47
        timestamp_col TIMESTAMP(3)
48
    ) WITH (
49
        'connector' = 'kafka',
50
        'topic' = 'my-topic',
51
        'properties.bootstrap.servers' = 'localhost:9092',
52
        'format' = 'avro'
53
    )
54
    """;
55
```
56

57
### Programmatic Schema Usage
58

59
```java { .api }
60
// Create deserializer for GenericRecord
61
Schema avroSchema = SchemaBuilder.record("MyRecord")
62
    .fields()
63
    .name("id").type().intType().noDefault()
64
    .name("name").type().stringType().noDefault()
65
    .endRecord();
66

67
AvroDeserializationSchema<GenericRecord> deserializer = 
68
    AvroDeserializationSchema.forGeneric(avroSchema);
69

70
// Create serializer for RowData
71
RowType rowType = RowType.of(
72
    new DataType[] {DataTypes.INT(), DataTypes.STRING()},
73
    new String[] {"id", "name"}
74
);
75

76
AvroRowDataSerializationSchema serializer = 
77
    new AvroRowDataSerializationSchema(rowType, AvroEncoding.BINARY, false);
78
```
79

80
## Architecture
81

82
### Core Components
83

84
The library provides several layers of Avro integration:
85

86
1. **Format Factories**: Auto-discovered services for SQL table format integration
87
2. **Schema Support**: Both generic records and specific typed records
88
3. **Row Data Integration**: Direct integration with Flink's internal row format
89
4. **File System Support**: Bulk readers/writers for Avro files
90
5. **Registry Integration**: Support for external schema registries
91

92
### Shaded Dependencies
93

94
All Apache Avro, Jackson, and Commons Compress dependencies are relocated to prevent classpath conflicts:
95
- `org.apache.avro.*` → `org.apache.flink.avro.shaded.org.apache.avro.*`
96
- `com.fasterxml.jackson.*` → `org.apache.flink.avro.shaded.com.fasterxml.jackson.*`
97
- `org.apache.commons.compress.*` → `org.apache.flink.avro.shaded.org.apache.commons.compress.*`
98

99
## Capabilities
100

101
### Configuration and Options
102

103
Configuration options for controlling Avro format behavior.
104

105
```java { .api }
106
public enum AvroEncoding {
107
    BINARY,  // Binary encoding for serialization/deserialization
108
    JSON     // JSON encoding for serialization/deserialization
109
}
110

111
public static final ConfigOption<String> AVRO_OUTPUT_CODEC;
112
public static final ConfigOption<AvroEncoding> AVRO_ENCODING;
113
public static final ConfigOption<Boolean> AVRO_TIMESTAMP_LEGACY_MAPPING;
114
```
115

116
[Configuration and Options](./configuration.md)
117

118
### Schema-based Serialization and Deserialization
119

120
Core functionality for converting between Java objects and Avro format.
121

122
```java { .api }
123
public static AvroDeserializationSchema<GenericRecord> forGeneric(Schema schema);
124
public static AvroSerializationSchema<GenericRecord> forGeneric(Schema schema);
125
public static <T extends SpecificRecord> AvroDeserializationSchema<T> forSpecific(Class<T> recordClazz);
126
public static <T extends SpecificRecord> AvroSerializationSchema<T> forSpecific(Class<T> recordClazz);
127
```
128

129
[Schema-based Serialization and Deserialization](./schemas.md)
130

131
### Row Data Integration
132

133
Integration with Flink's internal RowData format for table processing.
134

135
```java { .api }
136
public class AvroRowDataDeserializationSchema implements DeserializationSchema<RowData>;
137
public class AvroRowDataSerializationSchema implements SerializationSchema<RowData>;
138
```
139

140
[Row Data Integration](./rowdata.md)
141

142
### File System Operations
143

144
Bulk reading and writing of Avro files for batch processing.
145

146
```java { .api }
147
public class AvroInputFormat<T> extends FileInputFormat<T>;
148
public class AvroOutputFormat<T> extends FileOutputFormat<T>;
149
public class AvroBulkWriter<T> implements BulkWriter<T>;
150
public interface AvroBuilder<T>;
151
```
152

153
[File System Operations](./filesystem.md)
154

155
### Schema-Encoded Message Support
156

157
Support for messages with embedded schema information using configurable schema coders.
158

159
```java { .api }
160
public class RegistryAvroDeserializationSchema<T> extends AvroDeserializationSchema<T>;
161
public class RegistryAvroSerializationSchema<T> extends AvroSerializationSchema<T>;
162
public interface SchemaCoder;
163
```
164

165
[Schema-Encoded Message Support](./registry.md)
166

167
### Type System and Utilities
168

169
Utilities for converting between Flink types and Avro schemas.
170

171
```java { .api }
172
public class AvroSchemaConverter {
173
    public static Schema convertToSchema(LogicalType logicalType);
174
    public static DataType convertToDataType(String avroSchemaString);
175
    public static <T extends SpecificRecord> TypeInformation<Row> convertToTypeInfo(Class<T> avroClass);
176
}
177

178
public class AvroTypeInfo<T> extends TypeInformation<T>;
179
public class GenericRecordAvroTypeInfo extends AvroTypeInfo<GenericRecord>;
180
```
181

182
[Type System and Utilities](./utilities.md)