CREATE CHANGEFEED

On this page

The CREATE CHANGEFEED statement creates a new Enterprise changefeed, which targets an allowlist of tables, called "watched rows". Every change to a watched row is emitted as a record in a configurable format (JSON or Avro) to a configurable sink (Kafka or a cloud storage sink). You can create, pause, resume, or cancel an Enterprise changefeed.

For more information, see Stream Data Out of CockroachDB Using Changefeeds.

Required privileges

To create a changefeed, the user must be a member of the admin role or have the CREATECHANGEFEED parameter set.

Synopsis

Parameters

Sink URI

The sink URI follows the basic format of:

'{scheme}://{host}:{port}?{query_parameters}'

Kafka

Example of a Kafka sink URI:

'kafka://broker.address.com:9092?topic_prefix=bar_&tls_enabled=true&ca_cert=LS0tLS1CRUdJTiBDRVJUSUZ&sasl_enabled=true&sasl_user=petee&sasl_password=bones'

Cloud storage sink

Use a cloud storage sink to deliver changefeed data to OLAP or big data systems without requiring transport via Kafka.

Example of a cloud storage sink URI:

`experimental-s3://acme-co/employees?AWS_ACCESS_KEY_ID=123&AWS_SECRET_ACCESS_KEY=456`

Cloud storage sink URIs must be prepended with experimental- when working with changefeeds. For more information on the sink URI structure, see Use Cloud Storage for Bulk Operations.

Query parameters

Query parameters include:

Options

Avro limitations

Currently, support for Avro is limited and experimental. Below is a list of unsupported SQL types and values for Avro changefeeds:

• Decimals must have precision specified.
• INTERVAL, ARRAY, BIT, and collated STRING are not supported in Avro yet.

Avro types

Below is a mapping of CockroachDB types to Avro types:

Topic Naming

By default, a Kafka topic has the same name as the table that a changefeed was created on. If a changefeed was created on multiple tables, the changefeed will write to multiple topics corresponding to those table names. You can specify a topic prefix or use the full_table_name option to modify this.

You can either manually create a topic in your Kafka cluster before starting the changefeed, or the topic will be automatically created when the changefeed connects to your Kafka cluster.

Kafka has the following topic limitations:

• Legal characters are numbers, letters, and [._-].
• The maximum character length of a topic name is 249.
• Topics with a period (.) and underscore (_) can collide on internal Kafka data structures, so you should use either but not both.
• Characters not accepted by Kafka will be automatically encoded as unicode characters by CockroachDB.

Responses

Messages

The messages (i.e., keys and values) emitted to a sink are specific to the envelope. The default format is wrapped, and the output messages are composed of the following:

• Key: An array always composed of the row's PRIMARY KEY field(s) (e.g., [1] for JSON or {"id":{"long":1}} for Avro).
• Value:
  • One of three possible top-level fields:
    • after, which contains the state of the row after the update (or null' for DELETEs).
    • updated, which contains the updated timestamp.
    • resolved, which is emitted for records representing resolved timestamps. These records do not include an "after" value since they only function as checkpoints.
  • For INSERT and UPDATE, the current state of the row inserted or updated.
  • For DELETE, null.

For example:

Files

The files emitted to a sink use the following naming conventions:

• General file format
• Resolved file format

General file format

/[date]/[timestamp]-[uniquer]-[topic]-[schema-id]

For example:

/2020-04-02/202004022058072107140000000000000-56087568dba1e6b8-1-72-00000000-test_table-1.ndjson

Resolved file format

/[date]/[timestamp].RESOLVED

For example:

/2020-04-04/202004042351304139680000000000000.RESOLVED

Examples

Create a changefeed connected to Kafka

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka://host:port'
  WITH updated, resolved;

+--------------------+
|       job_id       |
+--------------------+
| 360645287206223873 |
+--------------------+
(1 row)

For more information on how to create a changefeed connected to Kafka, see Stream Data Out of CockroachDB Using Changefeeds.

Create a changefeed connected to Kafka using Avro

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka://host:port'
  WITH format = experimental_avro, confluent_schema_registry = <schema_registry_address>;

+--------------------+
|       job_id       |
+--------------------+
| 360645287206223873 |
+--------------------+
(1 row)

For more information on how to create a changefeed that emits an Avro record, see Stream Data Out of CockroachDB Using Changefeeds.

Create a changefeed connected to a cloud storage sink

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'experimental-scheme://host?parameters'
  WITH updated, resolved;

+--------------------+
|       job_id       |
+--------------------+
| 360645287206223873 |
+--------------------+
(1 row)

For more information on how to create a changefeed connected to a cloud storage sink, see Stream Data Out of CockroachDB Using Changefeeds.

Manage a changefeed

Use the following SQL statements to pause, resume, and cancel a changefeed.

Pause a changefeed

> PAUSE JOB job_id;

For more information, see PAUSE JOB.

Resume a paused changefeed

> RESUME JOB job_id;

For more information, see RESUME JOB.

Cancel a changefeed

> CANCEL JOB job_id;

For more information, see CANCEL JOB.

Configuring all changefeeds

It is useful to be able to pause all running changefeeds during troubleshooting, testing, or when a decrease in CPU load is needed.

To pause all running changefeeds:

PAUSE JOBS (SELECT job_id FROM [SHOW JOBS] WHERE job_type='CHANGEFEED' AND status IN ('running'));

This will change the status for each of the running changefeeds to paused, which can be verified with SHOW JOBS.

To resume all running changefeeds:

RESUME JOBS (SELECT job_id FROM [SHOW JOBS] WHERE job_type='CHANGEFEED' AND status IN ('paused'));

This will resume the changefeeds and update the status for each of the changefeeds to running.

Start a new changefeed where another ended

Find the high-water timestamp for the ended changefeed:

> SELECT * FROM crdb_internal.jobs WHERE job_id = <job_id>;

        job_id       |  job_type  | ... |      high_water_timestamp      | error | coordinator_id
+--------------------+------------+ ... +--------------------------------+-------+----------------+
  383870400694353921 | CHANGEFEED | ... | 1537279405671006870.0000000000 |       |              1
(1 row)

Use the high_water_timestamp to start the new changefeed:

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka//host:port'
  WITH cursor = '<high_water_timestamp>';

Note that because the cursor is provided, the initial scan is not performed.

See also

• Stream Data Out of CockroachDB Using Changefeeds
• Other SQL Statements
• Changefeed Dashboard

Was this helpful?