Learn Apache Flink - Source & Sink Integrations

#Introduction

The previous four episodes focused on processing inside Flink. Episode 8 opens the door: Flink is useless if it isn't connected to other systems. Modern streaming architectures are almost always hub-shaped — Kafka as the source, databases and object storage as destinations, and Flink in the middle doing transformations.

We'll connect Kafka, Kinesis, RabbitMQ, and file sources, then write results to Kafka, PostgreSQL, S3, and Elasticsearch. Finally, we'll discuss data formats — JSON, Avro, Protobuf, CSV — and the role of a schema registry in maintaining the contract between teams.

#File Sources and DataGen

#Reading Files and Object Storage

For data that's already stored, use FileSource. The path can point to a local filesystem or object storage like S3 or GCS:

import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
 
FileSource<String> fileSource = FileSource
    .forRecordStreamFormat(new TextLineInputFormat(), new Path("s3://bucket/logs/"))
    .monitorContinuously(Duration.ofMinutes(5))
    .build();

.monitorContinuously makes the source keep watching the folder for new files — turning "static" data into a living stream.

#DataGen for Testing

When the infrastructure isn't ready, use DataGen to generate synthetic data. It's a built-in connector that's very useful for experiments:

import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.api.common.typeinfo.Types;
 
DataGeneratorSource<String> gen = new DataGeneratorSource<>(
    index -> "event-" + index, 1000, Types.STRING);

DataGeneratorSource generates a thousand records without any external system — perfect for testing logic before integration.

#Kafka Source and Sink

#Kafka as the Main Source

Kafka is the most common source in the Flink ecosystem. Use KafkaSource with an appropriate deserializer:

import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.formats.json.JsonDeserializationSchema;
 
KafkaSource<Order> source = KafkaSource.<Order>builder()
    .setBootstrapServers("localhost:9092")
    .setTopics("orders")
    .setGroupId("flink-consumer")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new JsonDeserializationSchema<>(Order.class))
    .build();

OffsetsInitializer.earliest() starts reading from the beginning of the topic, while latest() only reads new data. This choice determines whether the job processes historical data or only starts from now.

#Writing to Kafka

The Kafka sink requires a serializer. Build it with KafkaSink and KafkaRecordSerializationSchema:

import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.formats.json.JsonSerializationSchema;
 
KafkaSink<Order> sink = KafkaSink.<Order>builder()
    .setBootstrapServers("localhost:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("processed-orders")
        .setValueSerializationSchema(new JsonSerializationSchema<Order>()).build())
    .build();
 
orders.sinkTo(sink);

KafkaSink supports exactly-once guarantees thanks to the two-phase commit mechanism with Kafka transactions — a great match for the checkpointing from episode 6.

#Spinning Up Kafka Locally

To experiment, spin up Kafka with Docker:

docker compose up -d kafka
docker compose logs -f kafka

Make sure the Kafka service is healthy before running a job. The docker compose up -d kafka command runs the container in the background, and docker compose logs -f kafka follows its logs in real-time.

#Sinking to Databases, Object Storage, and Elasticsearch

#JDBC to PostgreSQL

To write to a relational database, use JdbcSink:

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
 
orders.addSink(JdbcSink.sink(
    "INSERT INTO agg_orders (user_id, total) VALUES (?, ?)",
    (statement, order) -> {
        statement.setString(1, order.getUserId());
        statement.setLong(2, order.getAmount());
    },
    JdbcExecutionOptions.builder().withBatchSize(1000).build(),
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:postgresql://db:5432/flink")
        .withDriverName("org.postgresql.Driver")
        .withUsername("flink")
        .withPassword("secret")
        .build()));

JdbcSink.sink takes the SQL statement, a mapping function, execution options, and a connection. Store credentials in a secret manager rather than hardcoding them as in the example above.

#Other Sinks: Elasticsearch and Object Storage

• Elasticsearch: use ElasticsearchSink for real-time search and dashboards.
• Object storage: write with FileSink to S3 in Parquet or ORC format for a data lake.
• Other message queues: RabbitMQ and Kinesis each have their official connectors under flink-connector-*.

#Data Formats and Schema Management

#JSON, Avro, Protobuf, CSV

The data format determines how Flink turns bytes into objects:

CREATE TABLE orders (
  user_id STRING,
  amount  BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'format' = 'avro'
);

JSON is easy for humans to read, CSV is concise for files, and Avro and Protobuf are compact and schematized — suitable for production. With Avro, schemas are registered in the Schema Registry so producers and consumers always agree on the data contract.

#Choosing a Format

• JSON: fast for debugging, large overhead.
• Avro: compact, schematized, supports schema evolution.
• Protobuf: compact and strict, popular with teams using gRPC.
• CSV: simple for files and legacy integration.

ls $FLINK_HOME/lib | grep -i kafka

The ls $FLINK_HOME/lib | grep -i kafka command confirms the connector JAR is in the lib/ folder — a prerequisite for the standalone cluster to recognize the Kafka connector.

#Conclusion

Episode 8 connected Flink to the ecosystem: building KafkaSource and KafkaSink, using DataGen and FileSource, writing to PostgreSQL with JDBC, and choosing the JSON, Avro, Protobuf, and CSV data formats with clean schema management.

The key takeaways:

• KafkaSource and KafkaSink are the main bridge between Flink and the streaming world.
• Choose the starting read position (earliest or latest) according to the job's needs.
• JdbcSink, ElasticsearchSink, and FileSink cover most output needs.
• Avro and Protobuf win in production because they're schematized; JSON is for prototyping speed.
• Make sure the connector JAR is in lib/ before running a job on a standalone cluster.

In the next episode, episode 9, we'll discuss the Table API & SQL — writing Flink SQL queries, using TableEnvironment and catalogs, understanding temporal tables and CDC, and applying windowing, joins, and aggregations directly from SQL. This is the fastest path to Flink productivity.