- Release Notes and Announcements
- Product Introduction
- Purchase Guide
- Getting Started
- Operation Guide
- Instance Management
- Creating Instance
- Naming with Consecutive Numeric Suffixes or Designated Pattern String
- Viewing Instance
- Upgrading Instance
- Downgrading Instance Configuration
- Terminating/Returning Instances
- Change from Pay-as-You-Go to Monthly Subscription
- Upgrading Instance Version
- Adding Routing Policy
- Public Network Bandwidth Management
- Connecting to Prometheus
- AZ Migration
- Setting Maintenance Time
- Setting Message Size
- Topic Management
- Consumer Group
- Monitoring and Alarms
- Smart Ops
- Permission Management
- Tag Management
- Querying Message
- Event Center
- Migration to Cloud
- Data Compression
- Instance Management
- CKafka Connector
- Best Practices
- Connector Best Practices
- Connecting Flink to CKafka
- Connecting Schema Registry to CKafka
- Connecting Spark Streaming to CKafka
- Connecting Flume to CKafka
- Connecting Kafka Connect to CKafka
- Connecting Storm to CKafka
- Connecting Logstash to CKafka
- Connecting Filebeat to CKafka
- Multi-AZ Deployment
- Production and Consumption
- Log Access
- Replacing Supportive Route (Old)
- Troubleshooting
- API Documentation
- History
- Introduction
- API Category
- Making API Requests
- DataHub APIs
- ACL APIs
- Topic APIs
- BatchModifyGroupOffsets
- BatchModifyTopicAttributes
- CreateConsumer
- CreateDatahubTopic
- CreatePartition
- CreateTopic
- CreateTopicIpWhiteList
- DeleteTopic
- DeleteTopicIpWhiteList
- DescribeDatahubTopic
- DescribeTopic
- DescribeTopicAttributes
- DescribeTopicDetail
- DescribeTopicProduceConnection
- DescribeTopicSubscribeGroup
- FetchMessageByOffset
- FetchMessageListByOffset
- ModifyDatahubTopic
- ModifyTopicAttributes
- DescribeTopicSyncReplica
- Instance APIs
- Route APIs
- Other APIs
- Data Types
- Error Codes
- SDK Documentation
- General References
- FAQs
- Service Level Agreement (New Version)
- Contact Us
- Glossary
- Release Notes and Announcements
- Product Introduction
- Purchase Guide
- Getting Started
- Operation Guide
- Instance Management
- Creating Instance
- Naming with Consecutive Numeric Suffixes or Designated Pattern String
- Viewing Instance
- Upgrading Instance
- Downgrading Instance Configuration
- Terminating/Returning Instances
- Change from Pay-as-You-Go to Monthly Subscription
- Upgrading Instance Version
- Adding Routing Policy
- Public Network Bandwidth Management
- Connecting to Prometheus
- AZ Migration
- Setting Maintenance Time
- Setting Message Size
- Topic Management
- Consumer Group
- Monitoring and Alarms
- Smart Ops
- Permission Management
- Tag Management
- Querying Message
- Event Center
- Migration to Cloud
- Data Compression
- Instance Management
- CKafka Connector
- Best Practices
- Connector Best Practices
- Connecting Flink to CKafka
- Connecting Schema Registry to CKafka
- Connecting Spark Streaming to CKafka
- Connecting Flume to CKafka
- Connecting Kafka Connect to CKafka
- Connecting Storm to CKafka
- Connecting Logstash to CKafka
- Connecting Filebeat to CKafka
- Multi-AZ Deployment
- Production and Consumption
- Log Access
- Replacing Supportive Route (Old)
- Troubleshooting
- API Documentation
- History
- Introduction
- API Category
- Making API Requests
- DataHub APIs
- ACL APIs
- Topic APIs
- BatchModifyGroupOffsets
- BatchModifyTopicAttributes
- CreateConsumer
- CreateDatahubTopic
- CreatePartition
- CreateTopic
- CreateTopicIpWhiteList
- DeleteTopic
- DeleteTopicIpWhiteList
- DescribeDatahubTopic
- DescribeTopic
- DescribeTopicAttributes
- DescribeTopicDetail
- DescribeTopicProduceConnection
- DescribeTopicSubscribeGroup
- FetchMessageByOffset
- FetchMessageListByOffset
- ModifyDatahubTopic
- ModifyTopicAttributes
- DescribeTopicSyncReplica
- Instance APIs
- Route APIs
- Other APIs
- Data Types
- Error Codes
- SDK Documentation
- General References
- FAQs
- Service Level Agreement (New Version)
- Contact Us
- Glossary
Overview
CKafka Connector offers data distribution capabilities. You can distribute CKafka data to TDSQL-C for PostgreSQL for further storage, query, and analysis.
Prerequisites
Currently, this feature relies on the TDSQL-C for PostgreSQL service, which should be activated first.
Directions
1. Log in to the CKafka console.
2. Click Connector > Task Management > Task List on the left sidebar, select the region, and click Create Task.
3. Enter the task name, select Data Distribution as the Task Type, select TDSQL-C for PostgreSQL as the Data Target Type, and click Next.
4. Configure the data source information.
Source Topic: Select the data source topic.
Elastic Topic: Select the created elastic topic. For more information, see Topic Management.
CKafka Instance Topic: Select the created CKafka instance and topic. If the instance is configured with ACL policies, ensure that the selected topic has read/write permissions. For more information, see Creating Topic.
Start Offset: Select the topic offset of historical messages when dumping.
5. Click Next, click Preview Data, and the first message from the specified Source Topic will be obtained and parsed.
Note:
Currently, message parsing must meet the following requirements:
The message is a JSON string.
The source data must be in single-level JSON format. To convert nested JSON into single-level JSON, see Data Processing Rule Description.
6. (Optional) Enable Data Processing Rule. For more information, see Data Parsing.
7. Click Next to configure the data target.
Data Target: Select the created connection to TDSQL-C.
Database: Select the source database.
Table: Select the source table.
Database Sync Mode
Default field match: This option is only used for the following:
The source topic data is the binlog/row-level changes data (insertion, deletion, or modification) of a single table subscribed by CKafka Connector from MySQL/PostgreSQL;
The source topic data must have a primary key and contain a schema.
Field match one by one:
Source Data: Click to pull the source topic data. You need to select the matching fields in the target table for the message fields one by one.
Insertion Mode: This option supports INSERT or UPSERT. If you select UPSERT, you need to select the primary key (when the inserted rows conflict, the task will update the columns of the conflicting rows except the primary key).
Upstream Data Format: JSON and Debezium are supported.
Note:
When the upstream MySQL binlog/PostgreSQL row-level changes data table structure changes, the changes can be synced to the downstream PostgreSQL.
Handle Failed Message: Specify the method of handling failed messages. You can select Discard, Retain, or Deliver to CLS (for this option, you need to specify the target logset and log topic and grant the CLS access).
Retain: This mode is suitable for the test environment. When a task fails, it will be terminated without retry, and the cause of failure will be recorded in the event center.
Discard: This mode is suitable for the production environment. When a task fails, the current failure message will be ignored. We recommend you use the Retain mode to conduct a test first and then change the task to Discard mode for production.
Deliver to CLS: This mode is suitable for a strict production environment. When a task fails, the failure message, metadata, and cause of failure will be uploaded to the specified CLS topic.
8. Click Submit, and you can see the created task in the Task List. You can also check the task creation progress on the status bar.
Yes
No
Was this page helpful?