The CKafka Connector facilitates data outflow, enabling you to distribute CKafka data to PostgreSQL for subsequent storage, interrogation, and examination.
Preparations
Currently, this feature relies on the TencentDB for PostgreSQL service, which should be activated first.
2. Click Connector > Task List on the left sidebar, select the region, and click Create Task.
3. Enter the task name, select 'Data Outflow' for the task type, choose 'PostgreSQL' as the data target type, and then click 'Next'.
4. Configure the data source information.
Source Topic: Select the data source topic.
Elastic Topic: Select the pre-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.
Starting Position: Select the topic offset of historical messages when dumping.
5. After configuring the above information, click Next, click Preview Topic Message, 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 for simple message format conversion.
6. (Optional) Enable the Data Processing for Source Data button. For more information, see Simple Data Processing.
7. Click Next to configure the data target information.
Data Target: Select the pre-established connection to PostgreSQL.
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-by-field Matching:
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: Supports either insert or upsert. When choosing upsert, a primary key must be selected (if an inserted row causes a conflict, the task will update the remaining columns of the conflicting row, excluding the primary key).
Upstream Data Format: Data processing is currently not supported for the Debezium format.
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 on 'Submit'. The task you just created will appear in the task list, and the progress of task creation can be monitored in the status bar.
