AnalyticDB:Synchronize Kafka data to AnalyticDB for MySQL using data synchronization (APS)

Step 1: Add a Kafka data source

Skip this step if you have already added a Kafka data source. Go to Step 2: Create a sync link.

1. Log on to the AnalyticDB for MySQL console. In the upper-left corner, select a region. In the left-side navigation pane, click Clusters, and then click your cluster ID.

2. In the left-side navigation pane, choose Data Ingestion > Data Source Management.

3. In the upper-right corner, click Create Data Source.

4. Configure the following parameters, and then click Create.

   Parameter	Description
   Data source type	Select Kafka.
   Data source name	A name is auto-generated from the data source type and current time. Change it as needed.
   Data source description	A description of the data source, such as its use cases or business scope.
   Cloud provider	Only Alibaba Cloud instances are supported.
   Kafka instance	The ID of the Kafka instance. Log on to the ApsaraMQ for Kafka console and find the instance ID on the Clusters page.
   Kafka topic	The name of the topic. Log on to the ApsaraMQ for Kafka console and find the topic name on the Topic Management page of the target instance.
   Message data format	Only JSON is supported.

Step 2: Create a sync link

1. In the left-side navigation pane, click Simple Log Service/Kafka Data Synchronization.

2. In the upper-left corner, click Create Synchronization Job, and then select the Kafka Data Source tab.

3. In the Source and destination settings section, configure the following parameters:

   Parameter	Description
   Job name	A name is auto-generated from the data source type and current time. Change it as needed.
   Data source	Select an existing Kafka data source or create a new one.
   Data source format	Only JSON is supported.
   Destination type	Select Data Warehouse - AnalyticDB for MySQL Storage.
   AnalyticDB for MySQL account	The database account for the AnalyticDB for MySQL cluster.
   AnalyticDB for MySQL password	The password for the database account.

4. In the Destination database and table settings section, configure the following parameters:

   Parameter	Description
   Database name	The name of the destination database in the AnalyticDB for MySQL cluster.
   Table name	The name of the destination table.
   Sample data	The latest message from the Kafka topic is automatically retrieved and displayed here. Review this sample to verify the topic data is in the expected JSON format before proceeding.
   Parsed JSON layers	The number of nested JSON layers to parse. Valid values: 0 (no parsing), 1 (default, parses one layer), 2, 3, 4. For details on JSON nested parsing behavior, see Use the data synchronization feature (APS) to synchronize Kafka data.
   Schema field mapping	Displays the fields parsed from the sample data. Modify destination field names and types, or add and remove fields as needed.

5. In the Synchronization settings section, configure the following parameters:

   Parameter	Description
   Start offset	The point in time from which the task begins consuming Kafka data. The task starts from the first record at or after this time.
   Dirty data processing mode	How to handle records where the source data type does not match the destination field type. Stop Synchronization (default): the task stops on dirty data; update the destination field type or change this setting, then restart the task. Process as NULL: writes NULL to the mismatched field and continues writing the remaining fields normally. For example, if a row has three fields (col1, col2, col3) and col2 contains dirty data, col2 is written as NULL while col1 and col3 are written normally.
   Job resource group	The job resource group that the task runs in.
   ACUs for incremental synchronization	The number of ACUs allocated from the job resource group. Minimum: 2 ACUs. Maximum: the total available ACUs in the resource group. Allocate more ACUs to improve ingestion throughput and task stability. A sync task holds its allocated ACUs for the entire duration of the task, reducing the ACUs available to other tasks in the same resource group. For example, if the resource group has a maximum of 48 ACUs and one task uses 8 ACUs, subsequent tasks in that group can use at most 40 ACUs.
   Add to whitelist	Add the Kafka vSwitch CIDR block to the AnalyticDB for MySQL cluster whitelist to allow network connectivity for data synchronization.

6. Click Submit.

Step 3: Start the sync task

1. On the Simple Log Service/Kafka Data Synchronization page, find the task you created and click Start in the Actions column.

2. Click Search. When the task status changes to Running, the sync task has started successfully.