AnalyticDB:Synchronize Kafka data using the APS data synchronization feature (Recommended)

Limitations

• Only JSON-formatted Kafka messages are supported. Other formats cause synchronization errors.

• Kafka table schema changes are not propagated to AnalyticDB for MySQL automatically. You must make DDL changes manually.

• Sample Kafka data larger than 8 KB is truncated by the Kafka API, which prevents the system from parsing the sample and auto-generating field mappings.

• Data written to AnalyticDB for MySQL is not visible until a Commit operation runs. The default Commit interval is 5 minutes, so wait at least 5 minutes after starting a task before querying the first batch of data.

• OSS paths across synchronization tasks cannot share a prefix. For example, oss://testBucketName/test/sls1/ and oss://testBucketName/test/ conflict and will cause data to be overwritten.

• If a synchronization task fails and the Kafka topic data has expired, the cleared data cannot be recovered when you restart the task. To reduce this risk, increase the topic's data retention period. If a task fails, contact technical support promptly.

Prerequisites

Before you begin, make sure you have:

• An AnalyticDB for MySQL Enterprise Edition, Basic Edition, or Data Lakehouse Edition cluster

• A job resource group

• A database account for the cluster (see the table below)

• An ApsaraMQ for Kafka instance in the same region as the AnalyticDB for MySQL cluster

• A Kafka topic with messages already sent to it (see Quick start for ApsaraMQ for Kafka)

Database account requirements by account type:

Billing

Using the data synchronization feature incurs the following fees:

• AnalyticDB for MySQL: Elastic ACU resource fees. See Billing for Data Lakehouse Edition and Billing for Enterprise Edition and Basic Edition.

• Object Storage Service (OSS): Storage fees, GET request fees, and PUT and other request fees. See Billing overview.

How it works

1. Add a Kafka data source to identify the ApsaraMQ for Kafka instance and topic to read from.

2. Create a data link (synchronization job) that maps Kafka messages to an AnalyticDB for MySQL table, configures JSON parsing, partition keys, and the consumer offset.

3. Start the task. The task begins consuming Kafka data from the selected offset and writes it to the destination table.

4. Query the ingested data using Spark SQL.

Step 1: Create a data source

Skip this step if you have already added a Kafka data source. Go directly to Step 2: Create a data 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, then click the cluster ID.

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

3. Click Create Data Source.

4. On the Create Data Source page, configure the following parameters:

   Parameter	Description
   Data Source Type	Select Kafka.
   Data Source Name	Auto-generated from the source type and current time. Change it as needed.
   Data Source Description	Enter a description, such as the business scenario or data scope.
   Deployment Mode	Only Alibaba Cloud Instance is supported.
   Kafka Instance	The Kafka instance ID. Find it on the Instances page of the ApsaraMQ for Kafka console.
   Kafka Topic	The topic name. Find it on the Topics page of the destination instance in the ApsaraMQ for Kafka console.
   Message Data Format	Only JSON is supported.

5. Click Create.

Step 2: Create a data link

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

2. Click Create Synchronization Job.

3. On the Create Synchronization Job page, configure the three sections below.

Step 3: Start the data synchronization 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. The task has started successfully when its status changes to Running.

The first batch of data is queryable at least 5 minutes after the task starts, because data is committed in 5-minute intervals by default.

Step 4: Analyze the data

After data is synchronized, use Spark SQL to query it in AnalyticDB for MySQL. For more information, see Spark development editor and Offline Spark application development.

1. In the left-side navigation pane, choose Job Development > Spark JAR Development.

2. Enter your Spark SQL statements in the default template and click Run Now. The following is an example:

   -- Example of Spark SQL. Modify the content and run your Spark program.
   
   conf spark.driver.resourceSpec=medium;
   conf spark.executor.instances=2;
   conf spark.executor.resourceSpec=medium;
   conf spark.app.name=Spark SQL Test;
   conf spark.adb.connectors=oss;
   
   -- SQL statements
   show tables from lakehouse20220413156_adbTest;

3. (Optional) On the Applications tab, click Logs in the Actions column to view the Spark SQL run logs.

Step 5 (Optional): Manage the data source

Go to Data Ingestion > Data Sources. The following operations are available in the Actions column:

JSON parsing levels and schema inference

The Parsed JSON Layers setting controls how many nested levels of a JSON message are expanded into separate destination fields.

Example message:

{
  "name" : "zhangle",
  "age" : 18,
  "device" : {
    "os" : {
        "test":lag,
        "member":{
             "fa":zhangsan,
             "mo":limei
           }
         },
    "brand" : "none",
    "version" : "11.4.2"
  }
}

Level 0 — No parsing. The entire JSON message is output as a single field.

Level 1 (default) — Top-level fields are expanded.

Level 2 — Two levels expanded. Non-nested fields are output directly; nested fields expand to their sub-fields.

Level 3

Level 4

What's next

• Spark development editor

• Offline Spark application development

• Billing for Data Lakehouse Edition

• Billing for Enterprise Edition and Basic Edition