Flink SQL Client and Iceberg Integration Demo

Overview

This reference covers a basic integration scenario for Data Fabric Streams and Iceberg tables by using DEP Polaris. For additional information, see:

• Apache Flink: Table API & SQL
• Apache Iceberg: Flink integration
• Apache Polaris

Preparation

The following sections describe the preparation for the demo for sourcing and sinking data between Data Fabric Streams and Iceberg tables.

Data Fabric Streams topic creation

Create the stream and the necessary topics:

maprcli stream create -path /user/mapruser1/s
maprcli stream topic create -path /user/mapruser1/s -topic input_employees
maprcli stream topic create -path /user/mapruser1/s -topic output_employees            

Produce a sample record to the input topic:

echo '0,Alice,0' | /opt/mapr/kafka/kafka-3.6.1/bin/kafka-console-producer.sh --topic /user/mapruser1/s:input_employees            

Flink configuration

Configure the AWS_REGION, and set AWS_REQUEST_CHECKSUM_CALCULATION and AWS_RESPONSE_CHECKSUM_VALIDATION to WHEN_REQUIRED in both Job Manager and Task Managers:

containerized:
  master:
    env:
      AWS_REGION: "us-east-1"
      AWS_REQUEST_CHECKSUM_CALCULATION: WHEN_REQUIRED
      AWS_RESPONSE_CHECKSUM_VALIDATION: WHEN_REQUIRED
  taskmanager:
    env:
      AWS_REGION: "us-east-1"
      AWS_REQUEST_CHECKSUM_CALCULATION: WHEN_REQUIRED
      AWS_RESPONSE_CHECKSUM_VALIDATION: WHEN_REQUIRED

Enable checkpoints for Iceberg writes. For the demo purposes, you can set a short checkpoint interval to see results faster:

execution.checkpointing.interval: 10s
execution.checkpointing.dir: maprfs:///path/to/checkpoints/dir

Downloading libraries

Create a directory for extra libraries:

mkdir ~/user-libs

You would require the following JAR files for the demo:

• flink-csv: To work with CSV records
• flink-sql-connector-kafka: To source and sink data from and to Data Fabric Streams
• iceberg-flink-runtime: Already included in the mapr-flink package

  Download the missing jars to the directory:

 wget -P ~/user-libs \
 https://repository.mapr.com/maven/org/apache/flink/flink-sql-connector-kafka/4.0.0.0-dep-1000/flink-sql-connector-kafka-4.0.0.0-dep-1000.jar \
 https://repository.mapr.com/maven/org/apache/flink/flink-csv/2.0.0.0-dep-1010/flink-csv-2.0.0.0-dep-1010.jar

The following sections in the given sequence demonstrate the sourcing and sinking of data between Data Fabric Streams and Iceberg tables, by using Flink SQL client and Polaris catalog.

Starting a Flink cluster

This demo assumes a running Flink cluster. If you don’t have a Flink cluster, start a YARN session cluster:

/opt/mapr/flink/flink-2.0.0/bin/yarn-session.sh --detached        

Starting the SQL client

Start the Flink SQL client and provide it with the downloaded libraries:

/opt/mapr/flink/flink-2.0.0/bin/sql-client.sh -l "file://${HOME}/user-libs"      

Optional: Set the result mode to tableau:

SET 'sql-client.execution.result-mode' = 'tableau';    

Creating the Polaris catalog

Create a Polaris catalog:

CREATE CATALOG polaris WITH (
	'type' = 'iceberg',
	'catalog-type' = 'rest',
	'uri' = 'https://<polaris_host>:<polaris_port>/api/catalog',
	'credential' = '<polaris_client_id>:<polaris_client_secret>',
	'warehouse' = 's3_catalog',
	'header.X-Iceberg-Access-Delegation' = 'vended-credentials',
	'rest.auth.type' = 'oauth2',
	'scope' ='PRINCIPAL_ROLE:ALL'
);

Defining tables

Let us create the following 3 tables:

Source table to consume records from a Data Fabric Streams topic:

CREATE TABLE kafka_input_employees (
	id INT,
	name VARCHAR,
	department_id INT
) WITH (
	'connector' = 'kafka',
	'format' = 'csv',
	'topic' = '/user/mapruser1/s:input_employees',
	'properties.bootstrap.servers' = 'fake:9092',
	'properties.group.id' = 'group1',
	'scan.startup.mode' = 'earliest-offset'
);

A sink table to produce records to another Data Fabric Streams topic:

CREATE TABLE kafka_output_employees (
	id INT,
	name VARCHAR,
	department_id INT
) WITH (
	'connector' = 'kafka',
	'format' = 'csv',
	'topic' = '/user/mapruser1/s:output_employees',
	'properties.bootstrap.servers' = 'fake:9092'
);

A table representing an Iceberg table for both sourcing and sinking:

CREATE TABLE polaris.demospace.employees (
	id INT,
	name VARCHAR,
	department_id INT
);

Reading from a Data Fabric Streams topic

Query the source table to view records from the Data Fabric Streams topic:

SELECT * FROM kafka_input_employees;

This is a continuous query that runs in the foreground, until you press ^C.

Following is the sample result of the query:

+----+-------------+--------------------------------+---------------+
| op |          id |                           name | department_id |
+----+-------------+--------------------------------+---------------+
| +I |           0 |                          Alice |             0 |

Piping data from Data Fabric Streams to Iceberg

To pipe data from a DF Streams topic into an Iceberg table, use the INSERT INTO ... SELECT FROM query:

INSERT INTO polaris.demospace.employees 
	SELECT * FROM kafka_input_employees;       

This is a continuous query and it will go in background and run infinitely until stopped explicitly.

Reading from an Iceberg table

To verify the result, query the Iceberg table:

SELECT * FROM polaris.demospace.employees;

Following is the result:

+----+-------------+--------------------------------+---------------+
| op |          id |                           name | department_id |
+----+-------------+--------------------------------+---------------+
| +I |           0 |                          Alice |             0 |
+----+-------------+--------------------------------+---------------+

If you have DEP Spark integrated with the same DEP Polaris instance, you should also be able to query the table there:

spark.sql("SELECT * FROM spark_demo.demospace.employees").show()       

Following is the sample result:

+---+-----+-------------+
| id| name|department_id|
+---+-----+-------------+
|  0|Alice|            0|
+---+-----+-------------+           

Dumping data from Iceberg to Data Fabric Streams

Insert data from the Iceberg table into the Data Fabric Streams topic:

INSERT INTO kafka_output_employees
    SELECT * FROM polaris.demospace.employees;

Consuming the output topic

To verify the load, consume from the output topic:

/opt/mapr/kafka/kafka-3.6.1/bin/kafka-console-consumer.sh --from-beginning --topic /user/mapruser1/s:output_employees

The query output or result would be as follows:

0,Alice,0