Maps (a.k.a. associative arrays or dictionaries) are a widely used data structure across programming languages, and Flink SQL provides full-featured native support. In this tutorial, we will walk through how to define and populate map columns, provide examples to convert between maps and strings, and show how to aggregate by map value.

You can run through this tutorial locally with the Flink SQL Client against Flink and Kafka running in Docker, or with Confluent Cloud. Run through these steps to provision Kafka and Flink.

git clone git@github.com:confluentinc/tutorials.git
cd tutorials

docker compose -f ./docker/docker-compose-flinksql.yml up -d

docker exec -it flink-sql-client sql-client.sh

CREATE TABLE user_preferences (
    user_id INT,
    preferences MAP<STRING, STRING>
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_preferences',
    'properties.bootstrap.servers' = 'broker:9092',
    'scan.startup.mode' = 'earliest-offset',
    'key.format' = 'raw',
    'key.fields' = 'user_id',
    'value.format' = 'avro-confluent',
    'value.avro-confluent.url' = 'http://schema-registry:8081',
    'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE user_preferences (
    user_id INT,
    preferences MAP<STRING, STRING>
);

Given the user_preferences table defined in the previous section, let's first insert a couple of records. Note that, in order to insert into a MAP column, use the MAP[key1, value1, key2, value2, ...].

INSERT INTO user_preferences VALUES
    (0, MAP['lang', 'en-US', 'ui-mode', 'dark']),
    (1, MAP['lang', 'es-ES', 'ui-mode', 'light']),
    (2, MAP['lang', 'en-US', 'ui-mode', 'light']),
    (3, MAP['ui-mode', 'light']);

Let's query the user_preferences table:

SELECT * FROM user_preferences;

You will see that the preferences map is rendered as key=value pairs in curly braces:

user_id                    preferences
      0     {lang=en-US, ui-mode=dark}
      1    {lang=es-ES, ui-mode=light}
      2    {lang=en-US, ui-mode=light}
      3                {ui-mode=light}

We can use bracket notation to select specific values. For example, to select the user's language preference:

SELECT
    user_id,
    preferences['lang'] AS lang
FROM user_preferences;

If the specified key doesn't exist in the MAP column, then NULL is returned, which can be useful for filtering out or detecting incomplete records:

SELECT
    user_id
FROM user_preferences
WHERE preferences['lang'] IS NULL;

The STR_TO_MAP function takes a single string of key-value pairs along with optional delimiter strings and creates a MAP from it. For example:

INSERT INTO user_preferences VALUES
    (4, STR_TO_MAP('lang=en-GB,ui-mode=dark'));

To convert a map into a string, simply concatenate using the || operator, e.g.:

SELECT
    user_id,
    'language: ' || preferences['lang'] || ', UI Mode: ' || preferences['ui-mode'] AS prefs
FROM user_preferences;

Flink SQL supports aggregation by MAP values. Simply drill into the MAP in the GROUP BY expression:

SELECT
    preferences['ui-mode'] AS ui_mode,
    COUNT(*) AS ui_mode_count
FROM user_preferences
GROUP BY preferences['ui-mode'];

When you are finished, clean up the infrastructure used for this tutorial, either by deleting the environment that you created in Confluent Cloud, or, if running in Docker:

docker compose -f ./docker/docker-compose-flinksql.yml down