0
# Apache Flink File Connector
1
2
Apache Flink File Connector provides a unified data source and sink for reading and writing files in both batch and streaming modes. It supports various file formats through StreamFormat and BulkFormat interfaces, with features like continuous monitoring, file splitting, compression support, and distributed processing capabilities across different file systems and object stores.
3
4
## Package Information
5
6
- **Package Name**: flink-connector-files
7
- **Package Type**: maven
8
- **Language**: Java
9
- **Group ID**: org.apache.flink
10
- **Installation**: `<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId><version>2.1.0</version></dependency>`
11
12
## Core Imports
13
14
```java
15
import org.apache.flink.connector.file.src.FileSource;
16
import org.apache.flink.connector.file.sink.FileSink;
17
import org.apache.flink.connector.file.src.reader.StreamFormat;
18
import org.apache.flink.connector.file.src.reader.BulkFormat;
19
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
20
import org.apache.flink.core.fs.Path;
21
```
22
23
## Basic Usage
24
25
### Reading Files
26
27
```java
28
import org.apache.flink.connector.file.src.FileSource;
29
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
30
import org.apache.flink.core.fs.Path;
31
import java.time.Duration;
32
33
// Create a file source for reading text files
34
FileSource<String> source = FileSource
35
.forRecordStreamFormat(new TextLineInputFormat(), new Path("/path/to/input/files"))
36
.monitorContinuously(Duration.ofSeconds(10))
37
.build();
38
39
// Use in DataStream API
40
DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "file-source");
41
```
42
43
### Writing Files
44
45
```java
46
import org.apache.flink.connector.file.sink.FileSink;
47
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
48
import org.apache.flink.core.fs.Path;
49
50
// Create a file sink for writing text files
51
FileSink<String> sink = FileSink
52
.forRowFormat(new Path("/path/to/output"), new SimpleStringEncoder<String>("UTF-8"))
53
.withRollingPolicy(DefaultRollingPolicy.builder()
54
.withMaxPartSize(MemorySize.ofMebiBytes(128))
55
.withRolloverInterval(Duration.ofMinutes(15))
56
.build())
57
.build();
58
59
// Use in DataStream API
60
stream.sinkTo(sink);
61
```
62
63
## Architecture
64
65
The Flink File Connector is built around several key components:
66
67
- **File Sources**: `FileSource` provides unified reading with support for both streaming and batch modes
68
- **File Sinks**: `FileSink` handles writing with exactly-once semantics and file rolling
69
- **Format Interfaces**: `StreamFormat` for record-by-record reading, `BulkFormat` for batch-oriented reading
70
- **File Discovery**: `FileEnumerator` implementations discover files and create splits for parallel processing
71
- **Split Assignment**: `FileSplitAssigner` manages distribution of file splits to reader nodes with locality awareness
72
- **Compression Support**: Automatic decompression for common formats (.gz, .bz2, .xz, .deflate)
73
- **File Compaction**: Optional compaction system to merge small files for better performance
74
75
## Capabilities
76
77
### File Source Operations
78
79
Unified file reading with support for various formats, continuous monitoring, and distributed processing.
80
81
```java { .api }
82
public static <T> FileSourceBuilder<T> forRecordStreamFormat(
83
final StreamFormat<T> streamFormat, final Path... paths);
84
85
public static <T> FileSourceBuilder<T> forBulkFileFormat(
86
final BulkFormat<T, FileSourceSplit> bulkFormat, final Path... paths);
87
```
88
89
[File Sources](./file-sources.md)
90
91
### File Sink Operations
92
93
Unified file writing with exactly-once semantics, bucketing, rolling policies, and optional compaction.
94
95
```java { .api }
96
public static <IN> DefaultRowFormatBuilder<IN> forRowFormat(
97
final Path basePath, final Encoder<IN> encoder);
98
99
public static <IN> DefaultBulkFormatBuilder<IN> forBulkFormat(
100
final Path basePath, final BulkWriter.Factory<IN> bulkWriterFactory);
101
```
102
103
[File Sinks](./file-sinks.md)
104
105
### Stream Format Interface
106
107
Interface for record-by-record file reading with automatic compression support.
108
109
```java { .api }
110
@PublicEvolving
111
public interface StreamFormat<T> extends Serializable, ResultTypeQueryable<T> {
112
Reader<T> createReader(
113
Configuration config, FSDataInputStream stream, long fileLen, long splitEnd)
114
throws IOException;
115
116
Reader<T> restoreReader(
117
Configuration config,
118
FSDataInputStream stream,
119
long restoredOffset,
120
long fileLen,
121
long splitEnd)
122
throws IOException;
123
124
boolean isSplittable();
125
126
@Override
127
TypeInformation<T> getProducedType();
128
129
@PublicEvolving
130
interface Reader<T> extends Closeable {
131
@Nullable
132
T read() throws IOException;
133
134
@Override
135
void close() throws IOException;
136
137
@Nullable
138
default CheckpointedPosition getCheckpointedPosition() {
139
return null;
140
}
141
}
142
}
143
```
144
145
[Stream Formats](./stream-formats.md)
146
147
### Bulk Format Interface
148
149
Interface for batch-oriented reading optimized for formats like ORC and Parquet.
150
151
```java { .api }
152
@PublicEvolving
153
public interface BulkFormat<T, SplitT extends FileSourceSplit>
154
extends Serializable, ResultTypeQueryable<T> {
155
156
BulkFormat.Reader<T> createReader(Configuration config, SplitT split) throws IOException;
157
158
BulkFormat.Reader<T> restoreReader(Configuration config, SplitT split) throws IOException;
159
160
boolean isSplittable();
161
162
@Override
163
TypeInformation<T> getProducedType();
164
165
interface Reader<T> extends Closeable {
166
@Nullable
167
RecordIterator<T> readBatch() throws IOException;
168
169
@Override
170
void close() throws IOException;
171
}
172
173
interface RecordIterator<T> {
174
@Nullable
175
RecordAndPosition<T> next();
176
177
void releaseBatch();
178
}
179
}
180
```
181
182
[Bulk Formats](./bulk-formats.md)
183
184
### File Discovery and Enumeration
185
186
File discovery mechanisms for finding and splitting files across distributed storage systems.
187
188
```java { .api }
189
public interface FileEnumerator {
190
Collection<FileSourceSplit> enumerateSplits(Path[] paths, int minDesiredSplits) throws IOException;
191
192
interface Provider extends Serializable {
193
FileEnumerator create();
194
}
195
}
196
```
197
198
[File Enumeration](./file-enumeration.md)
199
200
### Split Assignment
201
202
Split assignment strategies for distributing file processing across nodes with locality awareness.
203
204
```java { .api }
205
public interface FileSplitAssigner {
206
Optional<FileSourceSplit> getNext(String hostname);
207
void addSplits(Collection<FileSourceSplit> splits);
208
Collection<FileSourceSplit> remainingSplits();
209
210
interface Provider extends Serializable {
211
FileSplitAssigner create(Collection<FileSourceSplit> splits);
212
}
213
}
214
```
215
216
[Split Assignment](./split-assignment.md)
217
218
### File Compaction
219
220
File compaction system for merging small files to improve performance and reduce metadata overhead.
221
222
```java { .api }
223
public interface FileCompactor {
224
void compact(List<Path> inputFiles, Path outputFile) throws Exception;
225
}
226
227
public interface FileCompactStrategy {
228
long getSizeThreshold();
229
int getNumCheckpointsBeforeCompaction();
230
int getNumCompactThreads();
231
}
232
```
233
234
[File Compaction](./file-compaction.md)
235
236
## Types
237
238
### Core Split and Position Types
239
240
```java { .api }
241
public class FileSourceSplit {
242
public FileSourceSplit(String id, Path path, long offset, long length,
243
long modificationTime, long fileSize, String... hostnames);
244
public Path path();
245
public long offset();
246
public long length();
247
public long fileModificationTime();
248
public String[] hostnames();
249
public CheckpointedPosition getReaderPosition();
250
public FileSourceSplit updateWithCheckpointedPosition(CheckpointedPosition position);
251
}
252
253
public class CheckpointedPosition {
254
public static final long NO_OFFSET = -1L;
255
256
public CheckpointedPosition(long offset, long recordsAfterOffset);
257
public long getOffset();
258
public long getRecordsAfterOffset();
259
}
260
```
261
262
### Configuration Types
263
264
```java { .api }
265
public class ContinuousEnumerationSettings {
266
public ContinuousEnumerationSettings(Duration discoveryInterval);
267
public Duration getDiscoveryInterval();
268
}
269
```