AnalyticDB:Synchronize Kafka data with APS (recommended)

Prerequisites

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

• A job resource group is created.

• A database account is created for the AnalyticDB for MySQL cluster.

  • If you use an Alibaba Cloud account, you need to only create a privileged account.

  • If you use a Resource Access Management (RAM) user, you must create a privileged account and a standard account and associate the standard account with the RAM user.

• You have created a ApsaraMQ for Kafka (Kafka) instance in the same region as the AnalyticDB for MySQL cluster.

• You have created a Kafka topic and sent messages to it. For more information, see Quick start for Message Queue for Apache Kafka.

Notes

• Only Kafka data in JSON format is supported.

• Data in a Kafka topic is automatically deleted when its retention period expires. If a data synchronization job fails, data may be deleted from the topic before the job is restarted, causing data loss. To prevent this, increase the topic's data retention period and contact technical support immediately if a job fails.

• The Kafka API truncates sample data from Kafka that exceeds 8 KB. This causes JSON parsing to fail, preventing the system from automatically generating the field mapping.

• Changes to the Kafka source table schema do not trigger automatic DDL updates and are not synchronized to AnalyticDB for MySQL.

• After data ingestion, a commit operation is required to make the data visible. To ensure job stability and optimal read/write performance, the AnalyticDB for MySQL data synchronization feature uses a default 5-minute commit interval. Consequently, after you create and start a data synchronization job, you must wait at least 5 minutes to see the first batch of data.

Billing

The following fees apply when you use the AnalyticDB for MySQL data migration feature to migrate data to Object Storage Service (OSS):

• AnalyticDB for MySQL elastic ACU resource fees. For details, see Billing for Data Lakehouse Edition and Billing for Enterprise Edition and Basic Edition.

• OSS storage fees, GET request fees, and PUT and other request fees. For details, see Billing overview.

Workflow

• Step 1: Create a data source.

• Step 2: Create a synchronization link.

• Step 3: Start the data synchronization task.

• Step 4: Analyze data.

• Step 5 (Optional): Manage data sources.

Create a data source

1. Log on to the AnalyticDB for MySQL console. In the upper-left corner of the console, select a region. In the left-side navigation pane, click Clusters. Find the cluster that you want to manage and click the cluster ID.

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

3. In the upper-left corner, 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	The system generates a name based on the data source type and the current time. You can change the name as needed.
   Data Source Description	A description of the data source, such as its use case or business constraints.
   Deployment Mode	Currently, only Alibaba Cloud Instance is supported.
   Kafka Instance	The ID of the Kafka instance. Log on to the ApsaraMQ for Kafka console. On the Instances page, view the instance ID.
   Kafka Topic	The name of your Kafka topic. Log on to the ApsaraMQ for Kafka console. On the Topics page of the target instance, view the topic name.
   Message Data Format	The data format for messages. Currently, only JSON is supported.

5. After you configure the parameters, click Create.

Create a synchronization job

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

2. In the upper-left corner, click Create Synchronization Job.

3. On the Create Synchronization Job page, configure the parameters in the Source and Destination Settings, Destination Database and Table Settings, and Synchronization Settings sections.

   • The following table describes the parameters for Source and Destination Settings:

     Parameter	Description
     Job Name	Enter a name for the synchronization job. By default, the system generates a name based on the data source and current time, which you can customize.
     Data Source	Select an existing Kafka data source, or create a new one.
     Destination Type	Select one of the following options: Data Lake - User OSS. Data Lake - AnalyticDB Lake Storage (recommended). Important If you select Data Lake - AnalyticDB Lake Storage, you must first enable the lake storage feature.
     ADB Lake Storage	The name of the lake storage for the lake data of AnalyticDB for MySQL. Select the destination lake storage from the drop-down list. If no lake storage is available, click Automatically Created to create one. Important This parameter is required only when Destination Type is set to Data Lake - AnalyticDB Lake Storage.
     OSS Path	AnalyticDB for MySQL data lake storage path in OSS. Important This parameter is required only when Destination Type is set to Data Lake - User OSS. The drop-down list displays all buckets in the same region as the AnalyticDB for MySQL cluster. You can select any of them. Plan the storage path carefully. You cannot change this path after the job is created. We recommend that you select an empty directory. The OSS path cannot be a prefix of the OSS path for another synchronization job, or vice versa. This prevents data from being overwritten. For example, if two synchronization jobs have the OSS paths oss://testBucketName/test/sls1/ and oss://testBucketName/test/, the paths have a prefix relationship, and data might be overwritten during synchronization.
     Storage Format	The format for storing the data. Supported options include: PAIMON. Important This format is supported only when Destination Type is set to Data Lake - User OSS. ICEBERG.

   • The following table describes the parameters for Destination Database and Table Settings:

     Parameter	Description
     Database Name	The name of the destination database in AnalyticDB for MySQL. If a database with this name does not exist, a new one is created. Otherwise, data is synchronized to the existing database. For information about naming conventions, see Limitations. Important In the Source and Destination Settings section, if Storage Format is set to PAIMON, an existing database must meet the following conditions. Otherwise, the synchronization job fails: The database must be an external database. The CREATE DATABASE statement must be CREATE EXTERNAL DATABASE <database_name>. The DBPROPERTIES clause in the CREATE DATABASE statement must include the catalog property, and the value of catalog must be paimon. The DBPROPERTIES clause must include the adb.paimon.warehouse property. Example: adb.paimon.warehouse=oss://testBucketName/aps/data. The DBPROPERTIES clause must include the LOCATION property, and you must append .db to the database name in the path. Otherwise, XIHE queries fail. Example: LOCATION='oss://testBucketName/aps/data/kafka_paimon_external_db.db/'. For the OSS path specified by LOCATION, the bucket and directory must exist. Otherwise, the database creation fails.
     Table Name	The name of the destination table in AnalyticDB for MySQL. If a table with this name does not exist, a new one is created. The job fails if a table with this name already exists. For information about naming conventions, see Limitations.
     Sample Data	The system automatically retrieves the latest data from the Kafka topic to use as sample data. Note The data in the Kafka topic must be in JSON format. If the data is in another format, an error occurs during data synchronization.
     Parsed JSON Layers	Set the number of nested JSON levels to parse. Valid values: 0: Do not parse. 1 (default): Parse one level. 2: Parse two levels. 3: Parse three levels. 4: Parse four levels. For more information about the JSON parsing strategy, see JSON parsing levels and schema inference examples.
     Schema Field Mapping	This section displays the schema inferred from the sample data. You can then modify destination field names and data types, or add and remove fields.
     Partition Key Settings	Set a partition key for the destination table. We recommend that you partition by log time or business logic to improve data ingestion and query performance. If you do not set this parameter, the destination table is not partitioned by default. You can format the destination partition key by time or by a specified partition field. To partition data by date and time, select a date/time field as the partition key field. For the formatting method, select time formatting, and then specify the source field format and the destination partition format. AnalyticDB for MySQL uses the source field format to identify the field's value and then converts it to the destination partition format. For example, if the source field is gmt_created with a value of 1711358834, the source field format is a timestamp with second precision, and the destination partition format is yyyyMMdd, the data is partitioned by 20240325. To partition data by field value, select Specify partition field as the formatting method.

   • The following table describes the parameters for Synchronization Settings:

     Parameter	Description
     Starting Consumer Offset for Incremental Synchronization	Specifies the starting point from which the job consumes data from Kafka. Valid options: Earliest offset (begin_cursor): Consume data starting from the earliest available time point in Kafka. Latest offset (end_cursor): Consume data starting from the latest available time point in Kafka. Custom offset: Select a point in time. The system starts consuming data from the first record whose timestamp is greater than or equal to the selected time point.
     Job Resource Group	Select the Job Resource Group where the job runs.
     ACUs for Incremental Synchronization	Specify the number of AnalyticDB Compute Units (ACUs) in the Job Resource Group for the job. The minimum value is 2. The maximum value is the number of available compute resources in the Job Resource Group. We recommend allocating more ACUs to improve ingestion performance and job stability. Note When you create a data synchronization job, the job uses elastic resources in the Job Resource Group. Data synchronization jobs occupy resources for a long time, so the system deducts the resources used by the job from the Job Resource Group. For example, if a Job Resource Group has a maximum of 48 ACUs and an existing synchronization job uses 8 ACUs, the maximum number of ACUs available for another job in the same Job Resource Group is 40.
     Advanced Settings	Provides advanced options to customize the synchronization job. Applying these settings requires assistance from technical support.

4. After configuring the parameters, click Submit.

Start the data synchronization task

1. On the Simple Log Service/Kafka Data Synchronization page, select the data synchronization task and click Start in the Actions column.

2. Click Search in the upper-left corner. The task has started successfully when its status changes to Running.

Data analysis

After the synchronization task completes successfully, you can use Spark Jar development to analyze the data in AnalyticDB for MySQL. For more information about Spark development, see Spark development editor and Spark offline application development.

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

2. In the default template, enter the sample statements and click Run Now.

   -- Here is just an example of SparkSQL. 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;
   
   -- Here are your sql statements
   show tables from lakehouse20220413156_adbTest;

3. Optional: On the Applications tab, click Logs in the Actions column to view the Spark SQL job's run log.

Manage data sources

JSON parsing and schema inference examples

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

These examples show the parsing results for levels 0 through 4.