CREATE SINK: Kafka

CREATE SINK connects Materialize to an external system you want to write data to, and provides details about how to encode that data.

To use a Kafka broker (and optionally a schema registry) as a sink, make sure that a connection that specifies access and authentication parameters to that broker already exists; otherwise, you first need to create a connection. Once created, a connection is reusable across multiple CREATE SINK and CREATE SOURCE statements.

Syntax

sink_definition

sink_format_spec

kafka_sink_connection

csr_connection

CONNECTION options

CSR CONNECTION options

DOC ON option syntax

The DOC ON option has special syntax, shown above, with the following mechanics:

• The KEY and VALUE options specify whether the comment applies to the key schema or the value schema. If neither KEY or VALUE is specified, the comment applies to both types of schemas.

• The TYPE clause names a SQL type or relation, e.g. my_app.point.

• The COLUMN clause names a column of a SQL type or relation, e.g. my_app.point.x.

See Avro schema documentation for details on how documentation comments are added to the generated Avro schemas.

WITH options

Formats

The FORMAT option controls the encoding of the message key and value that Materialize writes to Kafka.

Known limitation: Materialize does not permit specifying the key format independently from the value format.

Avro

Syntax: FORMAT AVRO

When using the Avro format, the value of each Kafka message is an Avro record containing a field for each column of the sink’s underlying relation. The names and ordering of the fields in the record match the names and ordering of the columns in the relation.

If the KEY option is specified, the key of each Kafka message is an Avro record containing a field for each key column, in the same order and with the same names.

If a column name is not a valid Avro name, Materialize adjusts the name according to the following rules:

• Replace all non-alphanumeric characters with underscores.
• If the name begins with a number, add an underscore at the start of the name.
• If the adjusted name is not unique, add the smallest number possible to the end of the name to make it unique.

For example, consider a table with two columns named col-a and col@a. Materialize will use the names col_a and col_a1, respectively, in the generated Avro schema.

When using a Confluent Schema Registry:

• Materialize will automatically publish Avro schemas for the key, if present, and the value to the registry.

• You can specify the fullnames for the Avro schemas Materialize generates using the AVRO KEY FULLNAME and AVRO VALUE FULLNAME syntax.

• You can automatically have nullable fields in the Avro schemas default to null by using the NULL DEFAULTS option.

• You can add doc fields to the Avro schemas.

SQL types are converted to Avro types according to the following conversion table:

If a SQL column is nullable, and its type converts to Avro type t according to the above table, the Avro type generated for that column will be ["null", t], since nullable fields are represented as unions in Avro.

In the case of a sink on a materialized view, Materialize may not be able to infer the non-nullability of columns in all cases, and will conservatively assume the columns are nullable, thus producing a union type as described above. If this is not desired, the materialized view may be created using non-null assertions.

Avro schema documentation

Materialize allows control over the doc attribute for record fields and types in the generated Avro schemas for the sink.

For the container record type (named row for the key schema and envelope for the value schema, unless overridden by the AVRO ... FULLNAME options), Materialize searches for documentation in the following locations, in order:

1. For the key schema, a KEY DOC ON TYPE option naming the sink’s underlying relation. For the value schema, a VALUE DOC ON TYPE option naming the sink’s underlying relation.
2. A comment on the sink’s underlying relation.

For record types within the container record type, Materialize searches for documentation in the following locations, in order:

1. For the key schema, a KEY DOC ON TYPE option naming the SQL type corresponding to the record type. For the value schema, a VALUE DOC ON TYPE option naming the SQL type corresponding to the record type.
2. A DOC ON TYPE option naming the SQL type corresponding to the record type.
3. A comment on the SQL type corresponding to the record type.

Similarly, for each field of each record type in the Avro schema, Materialize documentation in the following locations, in order:

1. For the key schema, a KEY DOC ON COLUMN option naming the SQL column corresponding to the field. For the value schema, a VALUE DOC ON COLUMN option naming the column corresponding to the field.
2. A DOC ON COLUMN option naming the SQL column corresponding to the field.
3. A comment on the SQL column corresponding to the field.

For each field or type, Materialize uses the documentation from the first location that exists. If no documentation is found for a given field or type, the doc attribute is omitted for that field or type.

JSON

Syntax: FORMAT JSON

When using the JSON format, the value of each Kafka message is a JSON object containing a field for each column of the sink’s underlying relation. The names and ordering of the fields in the record match the names and ordering of the columns in the relation.

If the KEY option is specified, the key of each Kafka message is a JSON object containing a field for each key column, in the same order and with the same names.

SQL values are converted to JSON values according to the following conversion table:

Envelopes

The sink’s envelope determines how changes to the sink’s underlying relation are mapped to Kafka messages.

There are two fundamental types of change events:

• An insertion event is the addition of a new row to the underlying relation.
• A deletion event is the removal of an existing row from the underlying relation.

When a KEY is specified, an insertion event and deletion event that occur at the same time are paired together into a single update event that contains both the old and new value for the given key.

Upsert

Syntax: ENVELOPE UPSERT

The upsert envelope:

• Requires that you specify a unique key for the sink’s underlying relation using the KEY option. See upsert key selection for details.
• For an insertion event, emits the row without additional decoration.
• For an update event, emits the new row without additional decoration. The old row is not emitted.
• For a deletion event, emits a message with a null value (i.e., a tombstone).

Consider using the upsert envelope if:

• You need to follow standard Kafka conventions for upsert semantics.
• You want to enable key-based compaction on the sink’s Kafka topic while retaining the most recent value for each key.

Debezium

Syntax: ENVELOPE DEBEZIUM

The Debezium envelope wraps each event in an object containing a before and after field to indicate whether the event was an insertion, deletion, or update event:

// Insertion event.
{"before": null, "after": {"field1": "val1", ...}}

// Deletion event.
{"before": {"field1": "val1", ...}, "after": null}

// Update event.
{"before": {"field1": "oldval1", ...}, "after": {"field1": "newval1", ...}}

Note that the sink will only produce update events if a KEY is specified.

Consider using the Debezium envelope if:

• You have downstream consumers that want update events to contain both the old and new value of the row.
• There is no natural KEY for the sink.

Features

Automatic topic creation

If the specified Kafka topic does not exist, Materialize will attempt to create it using the broker’s default number of partitions, default replication factor, default compaction policy, and default retention policy.

If the connection’s progress topic does not exist, Materialize will attempt to create it with a single partition, the broker’s default replication factor, compaction enabled, and both size- and time-based retention disabled.

To customize a topic’s configuration, manually create the topic in Kafka with the desired configuration (e.g., using the kafka-topics.sh tool) before running CREATE SINK. If you choose to do so, observe the following guidance:

Exactly-once processing

By default, Kafka sinks provide exactly-once processing guarantees, which ensures that messages are not duplicated or dropped in failure scenarios.

To achieve this, Materialize stores some internal metadata in an additional progress topic. This topic is shared among all sinks that use a particular Kafka connection. The name of the progress topic can be specified when creating a connection; otherwise, a default name of _materialize-progress-{REGION ID}-{CONNECTION ID} is used. In either case, Materialize will attempt to create the topic if it does not exist. The contents of this topic are not user-specified.

End-to-end exactly-once processing

Exactly-once semantics are an end-to-end property of a system, but Materialize only controls the initial produce step. To ensure end-to-end exactly-once message delivery, you should ensure that:

• The broker is configured with replication factor greater than 3, with unclean leader election disabled (unclean.leader.election.enable=false).
• All downstream consumers are configured to only read committed data (isolation.level=read_committed).
• The consumers' processing is idempotent, and offsets are only committed when processing is complete.

For more details, see the Kafka documentation.

Required permissions

The access control lists (ACLs) on the Kafka cluster must allow Materialize to perform the following operations on the following resources:

When using automatic topic creation, Materialize additionally requires access to the following operations:

Troubleshooting

Upsert key selection

The KEY that you specify for an upsert envelope sink must be a unique key of the sink’s underlying relation.

Materialize will attempt to validate the uniqueness of the specified key. If validation fails, you’ll receive an error message like one of the following:

ERROR:  upsert key could not be validated as unique
DETAIL: Materialize could not prove that the specified upsert envelope key
("col1") is a unique key of the underlying relation. There are no known
valid unique keys for the underlying relation.

ERROR:  upsert key could not be validated as unique
DETAIL: Materialize could not prove that the specified upsert envelope key
("col1") is a unique key of the underlying relation. The following keys
are known to be unique for the underlying relation:
  ("col2")
  ("col3", "col4")

The first error message indicates that Materialize could not prove the existence of any unique keys for the sink’s underlying relation. The second error message indicates that Materialize could prove that col2 and (col3, col4) were unique keys of the sink’s underlying relation, but could not provide the uniqueness of the specified upsert key of col1.

There are three ways to resolve this error:

• Change the sink to use one of the keys that Materialize determined to be unique, if such a key exists and has the appropriate semantics for your use case.

• Create a materialized view that deduplicates the input relation by the desired upsert key:

  -- For each row with the same key `k`, the `ORDER BY` clause ensures we
  -- keep the row with the largest value of `v`.
  CREATE MATERIALIZED VIEW deduped AS
  SELECT DISTINCT ON (k) v
  FROM original_input
  ORDER BY k, v DESC;
  
  -- Materialize can now prove that `k` is a unique key of `deduped`.
  CREATE SINK s IN CLUSTER my_io_cluster
  FROM deduped
  INTO KAFKA CONNECTION kafka_connection (TOPIC 't')
  KEY (k)
  FORMAT JSON ENVELOPE UPSERT;
  

  NOTE: Maintaining the deduped materialized view requires memory proportional to the number of records in original_input. Be sure to assign deduped to a cluster with adequate resources to handle your data volume.

• Use the NOT ENFORCED clause to disable Materialize’s validation of the key’s uniqueness:

  CREATE SINK s IN CLUSTER my_io_cluster
  FROM original_input
  INTO KAFKA CONNECTION kafka_connection (TOPIC 't')
  -- We have outside knowledge that `k` is a unique key of `original_input`, but
  -- Materialize cannot prove this, so we disable its key uniqueness check.
  KEY k NOT ENFORCED
  FORMAT JSON ENVELOPE UPSERT;
  

  You should only disable this verification if you have outside knowledge of the properties of your data that guarantees the uniqueness of the key you have specified.

  WARNING! If the key is not in fact unique, downstream consumers may not be able to correctly interpret the data in the topic, and Kafka key compaction may incorrectly garbage collect records from the topic.

Examples

Creating a connection

A connection describes how to connect and authenticate to an external system you want Materialize to write data to.

Once created, a connection is reusable across multiple CREATE SINK statements. For more details on creating connections, check the CREATE CONNECTION documentation page.

Broker

CREATE SECRET kafka_ssl_key AS '<BROKER_SSL_KEY>';
CREATE SECRET kafka_ssl_crt AS '<BROKER_SSL_CRT>';

CREATE CONNECTION kafka_connection TO KAFKA (
    BROKER 'rp-f00000bar.data.vectorized.cloud:30365',
    SSL KEY = SECRET kafka_ssl_key,
    SSL CERTIFICATE = SECRET kafka_ssl_crt
);

CREATE SECRET kafka_password AS '<BROKER_PASSWORD>';

CREATE CONNECTION kafka_connection TO KAFKA (
    BROKER 'unique-jellyfish-0000-kafka.upstash.io:9092',
    SASL MECHANISMS = 'SCRAM-SHA-256',
    SASL USERNAME = 'foo',
    SASL PASSWORD = SECRET kafka_password
);

Confluent Schema Registry

CREATE SECRET csr_ssl_crt AS '<CSR_SSL_CRT>';
CREATE SECRET csr_ssl_key AS '<CSR_SSL_KEY>';
CREATE SECRET csr_password AS '<CSR_PASSWORD>';

CREATE CONNECTION csr_ssl TO CONFLUENT SCHEMA REGISTRY (
    URL 'https://rp-f00000bar.data.vectorized.cloud:30993',
    SSL KEY = SECRET csr_ssl_key,
    SSL CERTIFICATE = SECRET csr_ssl_crt,
    USERNAME = 'foo',
    PASSWORD = SECRET csr_password
);

CREATE SECRET IF NOT EXISTS csr_username AS '<CSR_USERNAME>';
CREATE SECRET IF NOT EXISTS csr_password AS '<CSR_PASSWORD>';

CREATE CONNECTION csr_basic_http
  FOR CONFLUENT SCHEMA REGISTRY
  URL '<CONFLUENT_REGISTRY_URL>',
  USERNAME = SECRET csr_username,
  PASSWORD = SECRET csr_password;

Creating a sink

Upsert envelope

CREATE SINK avro_sink
  IN CLUSTER my_io_cluster
  FROM <source, table or mview>
  INTO KAFKA CONNECTION kafka_connection (TOPIC 'test_avro_topic')
  KEY (key_col)
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection
  ENVELOPE UPSERT;

CREATE SINK json_sink
  IN CLUSTER my_io_cluster
  FROM <source, table or mview>
  INTO KAFKA CONNECTION kafka_connection (TOPIC 'test_json_topic')
  KEY (key_col)
  FORMAT JSON
  ENVELOPE UPSERT;

Debezium envelope

CREATE SINK avro_sink
  IN CLUSTER my_io_cluster
  FROM <source, table or mview>
  INTO KAFKA CONNECTION kafka_connection (TOPIC 'test_avro_topic')
  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection
  ENVELOPE DEBEZIUM;

Documentation comments

Consider the following sink, docs_sink, built on top of a relation t with several SQL comments attached.

CREATE TABLE t (key int NOT NULL, value text NOT NULL);
COMMENT ON TABLE t IS 'SQL comment on t';
COMMENT ON COLUMN t.value IS 'SQL comment on t.value';

CREATE SINK docs_sink IN CLUSTER ...
FROM t
INTO KAFKA CONNECTION kafka_connection (TOPIC 'doc-commont-example')
KEY (key)
FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_connection (
    DOC ON TYPE t = 'Top-level comment for container record in both key and value schemas',
    KEY DOC ON COLUMN t.key = 'Comment on column only in key schema',
    VALUE DOC ON COLUMN t.key = 'Comment on column only in value schema'
)
ENVELOPE UPSERT;

When docs_sink is created, Materialize will publish the following Avro schemas to the Confluent Schema Registry:

• Key schema:

  {
    "type": "record",
    "name": "row",
    "doc": "Top-level comment for container record in both key and value schemas",
    "fields": [
      {
        "name": "key",
        "type": "int",
        "doc": "Comment on column only in key schema"
      }
    ]
  }
  

• Value schema:

  {
    "type": "record",
    "name": "envelope",
    "doc": "Top-level comment for container record in both key and value schemas",
    "fields": [
      {
        "name": "key",
        "type": "int",
        "doc": "Comment on column only in value schema"
      },
      {
        "name": "value",
        "type": "string",
        "doc": "SQL comment on t.value"
      }
    ]
  }
  

See Avro schema documentation for details about the rules by which Materialize attaches doc fields to records.

Related pages

• SHOW SINKS
• DROP SINK