CREATE SOURCE: Kafka

CREATE SOURCE connects Materialize to an external system you want to read data from, and provides details about how to decode and interpret that data.

This page describes how to connect Materialize to a Kafka broker to read data from individual topics.

Syntax

format_spec

key_constraint

with_options

WITH options

Supported formats

Key-value encoding

By default, the message key is decoded using the same format as the message value. However, you can set the key and value encodings explicitly using the KEY FORMAT ... VALUE FORMAT syntax.

Features

Handling upserts

To create a source that uses the standard key-value convention to support inserts, updates, and deletes within Materialize, you can use ENVELOPE UPSERT:

CREATE SOURCE current_predictions
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'events'
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'https://localhost:8081'
  ENVELOPE UPSERT;

Note that:

• Using this envelope is required to consume log compacted topics.

Defining primary keys

Primary keys are automatically inferred for Kafka sources using the UPSERT or DEBEZIUM envelopes. For other source configurations, you can manually define a column (or set of columns) as a primary key using the PRIMARY KEY (...) NOT ENFORCED syntax. This enables optimizations and constructs that rely on a key to be present when it cannot be inferred.

Using Debezium

Materialize provides a dedicated envelope (ENVELOPE DEBEZIUM) to decode Kafka messages produced by Debezium. To create a source that interprets Debezium messages:

CREATE SOURCE kafka_repl
  FROM KAFKA BROKER 'kafka:9092' TOPIC 'pg_repl.public.table1'
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'http://schema-registry:8081'
  ENVELOPE DEBEZIUM;

Note that:

• If log compaction is enabled for your Debezium topic, you must use ENVELOPE DEBEZIUM UPSERT.

Any materialized view defined on top of this source will be incrementally updated as new change events stream in through Kafka, as a result of INSERT, UPDATE and DELETE operations in the original database.

For more details and a step-by-step guide on using Kafka+Debezium for Change Data Capture (CDC), check out Using Debezium.

Exposing source metadata

In addition to the message value, Materialize can expose the message key, headers and other source metadata fields to SQL.

Key

The message key is exposed via the INCLUDE KEY option. Composite keys are also supported (#7645).

CREATE SOURCE kafka_metadata
  FROM 'localhost:9092' TOPIC 'data'
  KEY FORMAT TEXT
  VALUE FORMAT TEXT
  INCLUDE KEY AS renamed_id;

Note that:

• This option requires specifying the key and value encodings explicitly using the KEY FORMAT ... VALUE FORMAT syntax.

• The UPSERT envelope always includes keys.

• The DEBEZIUM envelope is incompatible with this option.

Headers

Message headers are exposed via the INCLUDE HEADERS option, and are included as a column (named headers by default) containing a list of (text, bytea) pairs.

CREATE SOURCE kafka_metadata
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'data'
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'https://localhost:8081'
  INCLUDE HEADERS
  ENVELOPE NONE;

To retrieve the headers in a message, you can unpack the value:

SELECT key,
       field1,
       field2,
       headers[1].value AS kafka_header
FROM mv_kafka_metadata;

  key  |  field1  |  field2  |  kafka_header
-------+----------+----------+----------------
  foo  |  fooval  |   1000   |     hvalue
  bar  |  barval  |   5000   |     <null>

, or lookup by key:

SELECT key,
       field1,
       field2,
       thekey,
       value
FROM (SELECT key,
             field1,
             field2,
             unnest(headers).key AS thekey,
             unnest(headers).value AS value
      FROM mv_kafka_metadata) AS km
WHERE thekey = 'kvalue';

  key  |  field1  |  field2  |  thekey  |  value
-------+----------+----------+----------+--------
  foo  |  fooval  |   1000   |  kvalue  |  hvalue

Note that:

• The DEBEZIUM envelope is incompatible with this option.

Partition, offset, timestamp

These metadata fields are exposed via the INCLUDE PARTITION, INCLUDE OFFSET and INCLUDE TIMESTAMP options.

CREATE SOURCE kafka_metadata
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'data'
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'https://localhost:8081'
  INCLUDE PARTITION, OFFSET, TIMESTAMP AS ts
  ENVELOPE NONE;

SELECT "offset" FROM kafka_metadata WHERE ts > '2021-01-01';

offset
------
15
14
13

Note that:

• Using the INCLUDE OFFSET option with Debezium requires UPSERT semantics.

Setting start offsets

To start consuming a Kafka stream from a specific offset, you can use the start_offset option.

CREATE MATERIALIZED SOURCE kafka_offset
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'data'
  -- Start reading from the earliest offset in the first partition,
  -- the second partition at 10, and the third partition at 100
  WITH (start_offset=[0,10,100])
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'https://localhost:8081';

Note that:

• If fewer offsets than partitions are provided, the remaining partitions will start at offset 0. This is true if you provide start_offset=1 or start_offset=[1, ...].
• If more offsets than partitions are provided, then any partitions added later will incorrectly be read from that offset. So, if you have a single partition, but you provide start_offset=[1,2], when you add the second partition you will miss the first 2 records of data.
• Using an offset with a source envelope that can supply updates or deletes requires that Materialize handle possibly nonsensical events (e.g. an update for a row that was never inserted). For that reason, starting at an offset requires either a NONE envelope or a (DEBEZIUM) UPSERT envelope.

Time-based offsets

It’s also possible to set a start offset based on Kafka timestamps, using the kafka_time_offset option. This approach sets the start offset for each available partition based on the Kafka timestamp and the source behaves as if start_offset was provided directly.

It’s important to note that kafka_time_offset is a property of the source: it will be calculated once at the time the CREATE SOURCE statement is issued. This means that the computed start offsets will be the same for all views depending on the source and stable across restarts.

If you need to limit the amount of data maintained as state after source creation, consider using temporal filters instead.

WITH options

Authentication

SSL

To connect to a Kafka broker that requires SSL authentication, use the provided WITH options.

CREATE MATERIALIZED SOURCE kafka_ssl
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'top-secret' WITH (
      security_protocol = 'SSL',
      ssl_key_location = '/secrets/materialized.key',
      ssl_certificate_location = '/secrets/materialized.crt',
      ssl_ca_location = '/secrets/ca.crt',
      ssl_key_password = 'mzmzmz'
  )
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'https://localhost:8081' WITH (
      ssl_key_location = '/secrets/materialized.key',
      ssl_certificate_location = '/secrets/materialized.crt',
      ssl_ca_location = '/secrets/ca.crt'
  );

SSL WITH options

Confluent Schema Registry SSL WITH options

SASL

To connect to a Kafka broker that requires SASL authentication, use the provided WITH options.

SASL/PLAIN

CREATE SOURCE kafka_sasl
  FROM KAFKA BROKER 'broker.tld:9092' TOPIC 'top-secret' WITH (
      security_protocol = 'SASL_SSL',
      sasl_mechanisms = 'PLAIN',
      sasl_username = '<BROKER_USERNAME>',
      sasl_password = '<BROKER_PASSWORD>'
  )
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'https://schema-registry.tld' WITH (
      username = '<SCHEMA_REGISTRY_USERNAME>',
      password = '<SCHEMA_REGISTRY_PASSWORD>'
  );

This is the configuration required to connect to Kafka brokers running on Confluent Cloud.

SASL/GSSAPI (Kerberos)

CREATE SOURCE kafka_kerberos
  FROM KAFKA BROKER 'broker.tld:9092' TOPIC 'tps-reports' WITH (
      security_protocol = 'sasl_plaintext',
      sasl_kerberos_keytab = '/secrets/materialized.keytab',
      sasl_kerberos_service_name = 'kafka',
      sasl_kerberos_principal = 'materialized@CI.MATERIALIZE.IO'
  )
  FORMAT AVRO USING SCHEMA FILE '/tps-reports-schema.json';

Note that:

• Materialize does not support Kerberos authentication for the Confluent Schema Registry.

SASL WITH options

Examples

Creating a source

CREATE SOURCE avro_source
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'data'
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY 'https://localhost:8081';

CREATE SOURCE json_source
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'data'
  FORMAT BYTES;

CREATE MATERIALIZED VIEW jsonified_kafka_source AS
  SELECT
    data->>'field1' AS field_1,
    data->>'field2' AS field_2,
    data->>'field3' AS field_3
  FROM (SELECT CONVERT_FROM(data, 'utf8')::jsonb AS data FROM json_source);

CREATE SOURCE proto_source
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'billing'
  WITH (cache = true)
  FORMAT PROTOBUF MESSAGE '.billing.Batch'
  USING SCHEMA FILE '[path to schema]';

CREATE SOURCE text_source
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'data'
  FORMAT TEXT
  USING SCHEMA FILE '/scratch/data.json'
  ENVELOPE UPSERT;

CREATE SOURCE csv_source (col_foo, col_bar, col_baz)
  FROM KAFKA BROKER 'localhost:9092' TOPIC 'data'
  FORMAT CSV WITH 3 COLUMNS;

Related pages

• CREATE SOURCE
• Using Debezium