The Pulsar Flink connector is an integration of Apache Pulsar and Apache Flink (data processing engine), which allows Flink to read and write data to and from Pulsar.

The Pulsar Flink connector consists of the Pulsar DataStream connector and the Pulsar SQL connector.

Pulsar DataStream connector

The Pulsar DataStream connector is maintained as a part of the official Flink library. For details about the Pulsar DataStream connector, see Flink's documentation.

Pulsar SQL connector

Currently, the Pulsar SQL connector is not delivered as a part of the official Flink binary. It is maintained by StreamNative.

The Pulsar SQL connector allows you to query data from or write data into Pulsar topics using simple SQL queries or the Flink Table API.

Dependencies

This table outlines the Maven dependency that is added to the Pulsar SQL connector.

<table> <tr> <th> Maven Dependency </th> <th> SQL JAR </th> </tr> <tr> <td>

<dependency>
    <groupId>io.streamnative.connectors</groupId>
    <artifactId>flink-sql-connector-pulsar</artifactId>
    <version>1.15.1.1</version>
</dependency>

</td> <td>

SQL JAR

</td> </tr> </table>

Pulsar SQL connector options

This table outlines options for the Pulsar SQL connector.

Besides these Pulsar SQL connector options, you can configure the underlying Pulsar DataStream connector using the connector options defined in Apache Pulsar Source Connector and [Apache Pulsar Sink Connector]([subscription type](Apache Pulsar Source Connector using the WITH clause.

Features

This section describes some significant features for the Pulsar SQL connector.

Available metadata

The connector metadata are the metadata specific to the external system in use. Flink SQL supports using metadata from the external system or their decoding/encoding formats as a table column.

This table outlines the Pulsar topic metadata that can be mapped by the Pulsar SQL connector to Flink table fields.

The extended CREATE TABLE example demonstrates the syntax for exposing publish_time, producer_name, and topic metadata fields.

CREATE TABLE user
(
  `publish_time`  TIMESTAMP(3) METADATA VIRTUAL FROM 'publish_time',
  `producer_name` STRING METADATA VIRTUAL,
  `topic`         STRING METADATA VIRTUAL,
  `uid`           BIGINT,
  `item_id`       BIGINT,
  `description`   STRING
) WITH (
    'connector' = 'pulsar',
    'topics' = 'persistent://public/default/user',
    'service-url' = 'pulsar://localhost:6650',
    'admin-url' = 'http://localhost:8080',
    'source.start.message-id' = 'earliest' ,
    'format' = 'json'
);

Key and value formats

You can serialize or deserialize the key and the value of a Pulsar message to or from raw bytes using one of the Flink official formats. For details about how to configure the key and value formats, see configure key and value formats.

Value format

If the key of a Pulsar message is absent or ignored, you should use the format option (a synonym for value.format) to designate a decoded/encoded format for the Pulsar message. For details, see the formats documentation. All format options are prefixed with the format identifier, such as json.ignore-parse-errors.

CREATE TABLE users
(
  `uid`         BIGINT,
  `item_id`     BIGINT,
  `description` STRING
) WITH (
    'connector' = 'pulsar',
    'topics' = 'persistent://public/default/users',
    'service-url' = 'pulsar://localhost:6650',
    ...
    'format' = 'json',
    'json.ignore-parse-errors' = 'true'
);

You can use the value format to deserialize a Pulsar message to the following Flink data types:

ROW<`uid` BIGINT, `item_id` BIGINT, `description` STRING>

Key format

The deserialized or serialized key of a Pulsar message can be mapped to a Flink table field. The key format includes the fields that are listed in the key.fields option. You can use the semicolon symbol (;) to separate multiple fields. Thus, the key format is configured with the following data type:

ROW<`uid` BIGINT, `item_id` BIGINT>

And all the key fields are excluded in the value format's data type:

ROW<`description` STRING>

Topic and partition discovery

The topics and topic-pattern options specify the topics and topic pattern for the Pulsar DataStream connector to consume or produce data.

• topics: a list of topics that are separated using the semicolon symbol (;) like topic-1;topic-2.

• topic-pattern: uses the regular expression to discover the matched topics. For example, if the topic-pattern is set to test-topic-[0-9], then all topics whose names match the specified regular expression (starting with test-topic- and ending with a single digit) will be subscribed to by the consumer when running a query job.

By default, the Pulsar DataStream source connector dynamically checks topic changes every 30 seconds, such as the number of topic partitions and topics. To disable automatic topic and partition discovery, you can set a negative value for the pulsar.source.partitionDiscoveryIntervalMs option. For details, see Pulsar DataStream connector documentation.

Starting position

You can use the source.start.message-id or source.start-publish-time option to designate the starting position for the Pulsar DataStream source connector.

You can configure one of the following values for the source.start.message-id option:

• earliest
• latest
• A message ID (in <ledgerId>:<entryId>:<partitionIndex> format)

You can configure the source.start.publish-time option with a Long type timestamp value.

The source.start.message-id and source.start-publish-time options are exclusive. If both options are configured, the validation will fail. By default, the Pulsar DataStream source connector uses the StartCursor.earliest() option.

Sink topic router

The Pulsar DataStream connector supports designating a topic router. In the Pulsar SQL connector, you can configure the sink topic router using the sink.topic-routing-mode or sink.custom-topic-router option.

• sink.topic-routing-mode: the built-in topic router implementation. Available values are round-robin and message-key-hash.
• sink.custom-topic-router: the full class name of the custom topic router. Make sure that the custom topic router implementation is set on the classpath.

The sink.topic-routing-mode and sink.custom-topic-router options are exclusive. If both options are configured, the validation will fail.

Data type mapping

Pulsar Schema describes how to deserialize data from and serialize data to a message stored in Pulsar topics. Flink SQL also has its own data types as well as the deserialization and serialization framework. The Pulsar SQL connector provides multiple choices to map a Pulsar schema to Flink data types to transfer data between Flink SQL and Pulsar.

Flink SQL uses formats to decode and encode data from external systems, such as flink-csvs, flink-json, and flink-avro. If you want to read a Pulsar topic with a predefined Pulsar schema (JSON, CSV, or Avro schema), you can use the related Flink format.

This table outlines the recommended Flink format for each Pulsar Schema. To use the avro and json format, you should add flink-avro and flink-json to your dependencies if they are not added yet. (By default, the JAR package of the Pulsar SQL connector is bundled with them).

PulsarCatalog

PulsarCatalog supports configuring the Pulsar cluster as the metadata storage for Flink tables.

Explicit tables and native tables

PulsarCatalog defines two different kinds of tables: explicit tables and native tables.

• An explicit table is a table explicitly created using the CREATE statements or table API. It is like the common use pattern in other SQL connectors. You can create a table and then query data from or write data to the table.
• A native table is automatically created by PulsarCatalog. PulsarCatalog scans all non-system topics in a Pulsar cluster, and then maps each topic into a Flink table without using the CREATE statements.

Explicit tables

PulsarCatalog uses the Pulsar SchemaInfo to store the Flink schema bytes for an explicit table.

For each explicit table, PulsarCatalog creates a placehoder topic under a preconfigured tenant in the Pulsar cluster. The default value for such a preconfigured tenant is named __flink_catalog, but you can use the catalog-tenant option to specify a different tenant name. The Flink database is then mapped to a namespace with the same name. Finally, the placehoder topic that saves the Flink table schema information is named as table_<FLINK_TABLE_NAME>.

For example, if you create a table users under the testdb database, then a topic named table_users is created under the __flink_catalog tenant in thetestdb namespace.

The table like table_users is called a placeholder topic, because these topics do not have any producer or consumer. Therefore, no data would flow in or out of these topics. You can use the schema information of such topics to store the Flink table metadata.

For the JSON format, the Flink table schema is serialized and then stored under the schema key. Other table options from CREATE TABLE xxx WITH () statements are stored under the properties key.

You can use the pulsar-admin CLI tool to retrieve the JSON format of a topic schema:

pulsar-admin schemas get persistent://<tenant>/<namespace>/<topic>

Native tables

A native table does not have any placeholder topic. Instead, PulsarCatalog maps the Pulsar schema of the native table to a Flink table schema when creating a SQL job.

This table outlines the Flink formats that are used for a native table's Pulsar schema.

Comparison

After a native table is created, you can query data from existing Pulsar topics. PulsarCatalog automatically reads the topic's schema and decides which decoded/encoded format to use. However, the native table does not support watermarks and primary keys. Therefore, you cannot use the native table to do window aggregation. A native table maps tenant/namespace to a database and the topic name to the table name.

To fully manage a table, you can use the explicit table to define watermark fields, specify metadata fields, and specify a custom format. The usage is similar to creating a Pulsar table in GenericInMemoryCatalog. You can bind an explicit table to a Pulsar topic and each Pulsar topic can be bound to multiple Flink tables (including the native table).

PulsarCatalog options

<table class="configuration table table-bordered"> <thead> <tr> <th class="text-left" style="width: 20%">Key</th> <th class="text-left" style="width: 15%">Default</th> <th class="text-left" style="width: 5%">Type</th> <th class="text-left" style="width: 55%">Description</th> <th class="text-left" style="width: 5%">Required</th> </tr> </thead> <tbody> <tr> <td><h5>catalog-admin-url</h5></td> <td style="word-wrap: break-word;">"http://localhost:8080"</td> <td>String</td> <td>(Required) The admin URL of the Pulsar cluster.</td> <td>Yes</td> </tr> <tr> <td><h5>catalog-auth-params</h5></td> <td style="word-wrap: break-word;">(none)</td> <td>String</td> <td>The authentication parameters for accessing the Pulsar cluster.</td> <td></td> </tr> <tr> <td><h5>catalog-auth-plugin</h5></td> <td style="word-wrap: break-word;">(none)</td> <td>String</td> <td>The name of the authentication plugin for accessing the Pulsar cluster.</td> <td></td> </tr> <tr> <td><h5>catalog-service-url</h5></td> <td style="word-wrap: break-word;">"pulsar://localhost:6650"</td> <td>String</td> <td>(Required) The service URL of the Pulsar cluster.</td> <td>Yes</td> </tr> <tr> <td><h5>catalog-tenant</h5></td> <td style="word-wrap: break-word;">"__flink_catalog"</td> <td>String</td> <td>The Pulsar tenant that stores all table information.</td> <td></td> </tr> <tr> <td><h5>default-database</h5></td> <td style="word-wrap: break-word;">"default"</td> <td>String</td> <td>The default database when using PulsarCatalog. It will be created if it does not exist.</td> <td></td> </tr> </tbody> </table>

Create PulsarCatalog

You can create PulsarCatalog using any of the following ways:

• SQL query statements
• Flink Table API

This example shows how to create PulsarCatalog using the SQL query statements.

CREATE
CATALOG pulsar
  WITH (
    'type' = 'pulsar-catalog',
    'catalog-admin-url' = '<ADMIN_URL>',
    'catalog-service-url' = '<SERVICE_URL>'
);

This example shows how to create PulsarCatalog using the Flink Table API.

Catalog pulsarCatalog = new PulsarCatalog();
tableEnv.registerCatalog("pulsar",pulsarCatalog);

Examples

This section lists some examples about the Pulsar SQL connector.

Create a table

This example shows how to use the Pulsar SQL connector to create a table in a JSON encoding/decoding format.

CREATE TABLE users
(
  `user_id`  BIGINT,
  `item_id`  BIGINT,
  `description` STRING,
  `ts`       TIMESTAMP(3) METADATA FROM 'timestamp'
) WITH (
    'connector' = 'pulsar',
    'topics' = 'persistent://public/default/users',
    'service-url' = 'pulsar://localhost:6650',
    'admin-url' = 'http://localhost:8080',
    'format' = 'json'
)

Configure key and value formats

This example shows how to specify and configure the key format and value formats. The format options are prefixed with either the key or value plus the format identifier.

CREATE TABLE users
(
  `uid`         BIGINT,
  `item_id`     BIGINT,
  `description` STRING
) WITH (
    'connector' = 'pulsar',
    'topics' = 'persistent://public/default/users',
    'service-url' = 'pulsar://localhost:6650',
    'admin-url' = 'http://localhost:8080',
    'key.format' = 'json',
    'key.json.ignore-parse-errors' = 'true',
    'key.fields' = 'uid;item_id',
    'value.format' = 'json',
    'value.json.fail-on-missing-field' = 'false'

Configure authentication parameters

This example shows how to specify and configure the authentication parameters used by the pulsar-admin client.

CREATE TABLE users
(
  `uid`         BIGINT,
  `item_id`     BIGINT,
  `description` STRING
) WITH (
    'connector' = 'pulsar',
    'topics' = 'persistent://public/default/users',
    'service-url' = 'pulsar://localhost:6650',
    'admin-url' = 'http://localhost:8080',
    'pulsar.client.tlsAllowInsecureConnection' = 'true', 
    'pulsar.client.authPluginClassName' = 'org.apache.pulsar.client.impl.auth.AuthenticationToken',
    'pulsar.client.authParams' = 'token:eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ1c2VyMSJ9.2AgtxHe8-2QBV529B5DrRtpuqP6RJjrk21Mhn'

Relationship between the apache/flink repository and streamnative/flink repository

StreamNative forks the apache/flink repository and maintains the project in the streamnative/flink repository. Compared with the official Flink repository, the StreamNative forked repository maintains a different Pulsar DataStream connector and Pulsar SQL connector.

Currently, the apache/flink repository only contains the Pulsar DataStream connector while the streamnative/flink repository contains both the Pulsar DataStream connector and the Pulsar SQL connector.

• Pulsar DataStream connector: the code and distribution in the streamnative/flink repository differ slightly from those in the apache/flink repository. Usually, new features and bug fixes will first go into the streamnative/flink repository, and then related PRs are submitted to the apache/flink repository.
• Pulsar SQL connector: StreamNative will merge the Pulsar SQL connector back to the Flink community in the coming months.

For detailed differences and recommendations, see the following sections.

Components

Release process, version, and documentation

Issue and Feature tracking

• For the Pulsar DataStream connector, use Flink JIRA to submit bug reports and feature requests.
• For the Pulsar SQL connector, since it's not merged back to the official Flink repository yet, create an issue in the streamnative/flink repository to submit bug reports and feature requests. Once the Pulsar SQL connector is merged back to the official Flink repository, use the Flink JIRA as well.

Artifact locations