Flink Integration Overview

Get started with Apache Iceberg and Apache Flink for streaming and batch analytics

Apache Iceberg supports both Apache Flink’s DataStream API and Table API for streaming and batch analytics.

For a general setup path, start with the Quickstart

Feature Support

Iceberg provides comprehensive Flink integration with both SQL and DataStream APIs:

Feature	Support	Notes
SQL create catalog	✔️
SQL create database	✔️
SQL create table	✔️
SQL create table like	✔️
SQL alter table	✔️	Only support altering table properties, column and partition changes are not supported
SQL drop table	✔️
SQL select	✔️	Support both streaming and batch mode
SQL insert into	✔️	Support both streaming and batch mode
SQL insert overwrite	✔️
DataStream read	✔️
DataStream append	✔️
DataStream overwrite	✔️
Metadata tables	✔️
Rewrite files action	✔️

Getting Started

Prerequisites

Download Flink from the Apache download page. Iceberg uses Scala 2.12 when compiling the Apache iceberg-flink-runtime jar, so it’s recommended to use Flink bundled with Scala 2.12.

Download Flink

FLINK_VERSION=1.17.0
SCALA_VERSION=2.12
APACHE_FLINK_URL=https://archive.apache.org/dist/flink/
wget ${APACHE_FLINK_URL}/flink-${FLINK_VERSION}/flink-${FLINK_VERSION}-bin-scala_${SCALA_VERSION}.tgz
tar xzvf flink-${FLINK_VERSION}-bin-scala_${SCALA_VERSION}.tgz

# HADOOP_HOME is your hadoop root directory after unpack the binary package.
HADOOP_VERSION=2.8.5
APACHE_HADOOP_URL=https://archive.apache.org/dist/hadoop/
wget ${APACHE_HADOOP_URL}/common/hadoop-${HADOOP_VERSION}/hadoop-${HADOOP_VERSION}.tar.gz
tar xzvf hadoop-${HADOOP_VERSION}.tar.gz
HADOOP_HOME=`pwd`/hadoop-${HADOOP_VERSION}

export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

# Start the flink standalone cluster
cd flink-${FLINK_VERSION}/
./bin/start-cluster.sh

Flink SQL Client

Start the Flink SQL client with the Iceberg runtime jar:

# HADOOP_HOME is your hadoop root directory after unpack the binary package.
export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

# Below works for 1.15 or less
./bin/sql-client.sh embedded -j <flink-runtime-directory>/iceberg-flink-runtime-1.15-1.4.0.jar shell

# 1.16 or above has a regression in loading external jar via -j option.
# Put iceberg-flink-runtime-1.16-1.4.0.jar in flink/lib dir
./bin/sql-client.sh embedded shell

By default, Iceberg ships with Hadoop jars for Hadoop catalog. To use Hive catalog, load the Hive jars when opening the Flink SQL client.

Flink Python API

PyFlink 1.6.1 does not work on OSX with a M1 cpu

Install the Apache Flink dependency using pip:

pip install apache-flink==1.17.0

Add the Iceberg runtime jar:

import os
from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
iceberg_flink_runtime_jar = os.path.join(os.getcwd(), "iceberg-flink-runtime-1.17-1.4.0.jar")

env.add_jars("file://{}".format(iceberg_flink_runtime_jar))

Create a StreamTableEnvironment and execute Flink SQL:

from pyflink.table import StreamTableEnvironment

table_env = StreamTableEnvironment.create(env)
table_env.execute_sql("""
CREATE CATALOG my_catalog WITH (
    'type'='iceberg',
    'catalog-impl'='com.my.custom.CatalogImpl',
    'my-additional-catalog-config'='my-value'
)
""")

Run a query:

(table_env
    .sql_query("SELECT PULocationID, DOLocationID, passenger_count FROM my_catalog.nyc.taxis LIMIT 5")
    .execute()
    .print())

Type Conversion

Iceberg’s integration for Flink automatically converts between Flink and Iceberg types.

Flink to Iceberg

Flink	Iceberg	Notes
boolean	boolean
tinyint	integer
smallint	integer
integer	integer
bigint	long
float	float
double	double
char	string
varchar	string
string	string
binary	binary
varbinary	fixed
decimal	decimal
date	date
time	time
timestamp	timestamp without timezone
timestamp_ltz	timestamp with timezone
array	list
map	map
multiset	map
row	struct
raw		Not supported
interval		Not supported
structured		Not supported
timestamp with zone		Not supported
distinct		Not supported
null		Not supported
symbol		Not supported
logical		Not supported

Iceberg to Flink

Iceberg	Flink	Notes
boolean	boolean
struct	row
list	array
map	map
integer	integer
long	bigint
float	float
double	double
date	date
time	time
timestamp without timezone	timestamp(6)
timestamp with timezone	timestamp_ltz(6)
string	varchar(2147483647)
uuid	binary(16)
fixed(N)	binary(N)
binary	varbinary(2147483647)
decimal(P, S)	decimal(P, S)
nanosecond timestamp	timestamp(9)
nanosecond timestamp with timezone	timestamp_ltz(9)
unknown	null
variant		Not supported
geometry		Not supported
geography		Not supported

Next Steps

Connector Setup

Configure Flink connector for Iceberg tables

DDL Operations

Create and manage Iceberg tables with Flink DDL

Queries

Read data from Iceberg tables using Flink

Writes

Write data to Iceberg tables with Flink