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, Google Cloud Pub/Sub, a cloud storage sink, or a webhook sink). You can create, pause, resume, alter, or cancel an Enterprise changefeed.

We recommend reading the Changefeed Messages page for detail on understanding how changefeeds emit messages and Create and Configure Changefeeds for important usage considerations.

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={sasl user}&sasl_password={url-encoded password}&sasl_mechanism=SCRAM-SHA-256'

Google Cloud Pub/Sub

New in v22.1: Example of a Google Cloud Pub/Sub sink URI:

'gcpubsub://{project name}?region={region}&topic_name={topic name}&AUTH=specified&CREDENTIALS={base64-encoded key}'

Use Cloud Storage for Bulk Operations explains the requirements for the authentication parameter with specified or implicit. See Changefeed Sinks for further consideration.

Cloud Storage

The following are example file URLs for each of the cloud storage schemes:

Use Cloud Storage for Bulk Operations explains the requirements for authentication and encryption for each supported cloud storage sink. See Changefeed Sinks for considerations when using cloud storage.

Webhook

Example of a webhook URI:

'webhook-https://{your-webhook-endpoint}?insecure_tls_skip_verify=true'

See Changefeed Sinks for specifics on webhook sink configuration.

Query parameters

Query parameters include:

Options

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

New in v22.1: When emitting changefeed messages to a cloud storage sink, you can specify a partition format for your files using the partition_format query parameter. This will result in the following file path formats:

• daily: This is the default option and will follow the same pattern as the previous general file format.

• hourly: This will partition into an hourly directory as the changefeed emits messages, like the following:

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

• flat: This will result in no file partitioning. The cloud storage path you specify when creating a changefeed will store all of the message files with no additional directories created.

Resolved file format

/[date]/[timestamp].RESOLVED

For example:

/2020-04-04/202004042351304139680000000000000.RESOLVED

Examples

Before running any of the examples in this section it is necessary to enable the kv.rangefeed.enabled cluster setting. If you are working on a CockroachDB Serverless cluster, this cluster setting is enabled by default.

The following examples show the syntax for managing changefeeds and starting changefeeds to specific sinks. The Options table on this page provides a list of all the available options. For information on sink-specific query parameters and configurations see the Changefeed Sinks page.

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 step-by-step guidance on creating a changefeed connected to Kafka, see the Create a changefeed connected to Kafka example. The parameters table on the Changefeed Sinks page provides a list of all kafka-specific query parameters.

Create a changefeed connected to Kafka using Avro

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'kafka://host:port'
  WITH format = 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 this step-by-step example. The parameters table on the Changefeed Sinks page provides a list of all kafka-specific query parameters.

Create a changefeed connected to a cloud storage sink

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

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

For step-by-step guidance on creating a changefeed connected to a cloud storage sink, see the Changefeed Examples page. The parameters table on the Changefeed Sinks page provides a list of the available cloud storage parameters.

Create a changefeed with an S3 storage class

New in v22.1: To associate the changefeed message files with a specific storage class in your Amazon S3 bucket, use the S3_STORAGE_CLASS parameter with the class. For example, the following S3 connection URI specifies the INTELLIGENT_TIERING storage class:

CREATE CHANGEFEED FOR TABLE name INTO 's3://{BUCKET NAME}?AWS_ACCESS_KEY_ID={KEY ID}&AWS_SECRET_ACCESS_KEY={SECRET ACCESS KEY}&S3_STORAGE_CLASS=INTELLIGENT_TIERING' WITH resolved;

Use the parameter to set one of these storage classes listed in Amazon's documentation. For more general usage information, see Amazon's Using Amazon S3 storage classes documentation.

Create a changefeed connected to a Google Cloud Pub/Sub

> CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'gcpubsub://project name?parameters'
  WITH resolved;

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

For step-by-step guidance on creating a changefeed connected to a Google Cloud Pub/Sub, see the Changefeed Examples page. The parameters table on the Changefeed Sinks page provides a list of the available Google Cloud Pub/Sub parameters.

Create a changefeed connected to a webhook sink

CREATE CHANGEFEED FOR TABLE name, name2, name3
  INTO 'webhook-https://{your-webhook-endpoint}?insecure_tls_skip_verify=true'
  WITH updated;

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

For step-by-step guidance on creating a changefeed connected to a webhook sink, see the Changefeed Examples page. The parameters table on the Changefeed Sinks page provides a list of the available webhook parameters.

Manage a changefeed

For Enterprise changefeeds, use SHOW CHANGEFEED JOBS to check the status of your changefeed jobs:

> SHOW CHANGEFEED JOBS;

Use the following SQL statements to pause, resume, or 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.

Modify a changefeed

To modify an Enterprise changefeed, pause the job and then use:

ALTER CHANGEFEED job_id {ADD table DROP table SET option UNSET option};

You can add new table targets, remove them, set new changefeed options, and unset them.

For more information, see ALTER CHANGEFEED.

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 (WITH x AS (SHOW CHANGEFEED JOBS) SELECT job_id FROM x WHERE status = ('running'));

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

To resume all running changefeeds:

RESUME JOBS (WITH x AS (SHOW CHANGEFEED JOBS) SELECT job_id FROM x WHERE status = ('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

• Change Data Capture Overview
• SQL Statements
• Changefeed Dashboard
• Monitor and Debug Changefeeds