Skip to main content

Quick Start

This tutorial runs a streaming word count on a local Flink cluster, stores the results in Paimon, and reads the table in batch and streaming modes.

Jars​

Download and install the matching connector and dependencies using Installation. The tutorial uses a single-machine warehouse at file:/tmp/paimon.

Start​

Configure enough slots to run a writer and a reader at the same time. Set the following in your Flink configuration file (conf/config.yaml or, for older distributions, conf/flink-conf.yaml):

taskmanager.numberOfTaskSlots: 2

Start the cluster and SQL client:

<FLINK_HOME>/bin/start-cluster.sh
<FLINK_HOME>/bin/sql-client.sh

Check the Flink dashboard, then set each tutorial job to parallelism 1:

SET 'parallelism.default' = '1';

2. Create a Catalog and Table​

CREATE CATALOG my_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'file:/tmp/paimon'
);

USE CATALOG my_catalog;

CREATE TABLE word_count (
    word STRING PRIMARY KEY NOT ENFORCED,
    cnt BIGINT
);

The primary key lets Paimon update the count for each word. For catalog alternatives, see SQL DDL.

3. Write a Stream​

CREATE TEMPORARY TABLE word_table (
    word STRING
) WITH (
    'connector' = 'datagen',
    'fields.word.length' = '1',
    'rows-per-second' = '100'
);

SET 'execution.runtime-mode' = 'streaming';
SET 'execution.checkpointing.interval' = '10 s';
SET 'table.dml-sync' = 'false';

INSERT INTO word_count
SELECT word, COUNT(*) FROM word_table GROUP BY word;

Keep this job running. Wait for its first successful checkpoint in the Flink dashboard before querying: streaming writes become visible through committed Paimon snapshots.

4. Read the Current Results​

SET 'sql-client.execution.result-mode' = 'tableau';
SET 'execution.runtime-mode' = 'batch';
RESET 'execution.checkpointing.interval';

SELECT * FROM word_count;

Each batch query reads a snapshot and finishes. Repeat it to see updated counts. Switching the SQL client's runtime mode does not stop the streaming writer submitted in the previous step.

5. Follow Changes​

SET 'execution.runtime-mode' = 'streaming';

SELECT * FROM word_count;

The streaming query reads the current table and continues following changes. The result is unbounded; cancel it with Ctrl+C when you finish. See SQL Query for startup modes, time travel, and incremental reads.

6. Clean Up​

Cancel the remaining writer in the Flink dashboard, then exit the SQL client:

-- Optional: delete the tutorial table and its files after stopping the jobs.
-- DROP TABLE word_count;
EXIT;
<FLINK_HOME>/bin/stop-cluster.sh

For writer memory configuration, see Runtime Configuration.

Setting dynamic options​

For SQL hints and session options, see Runtime Configuration.

Next Steps​

TaskGuide
Choose keys, partitions, and a catalogSQL DDL
Insert, overwrite, update, or delete dataSQL Write
Enrich a stream with dimension dataLookup Joins
Ingest changes from an external databaseCDC Ingestion
Build a job in JavaFlink API