Build a real-time data lake by using DLA and DTS to synchronize data from ApsaraDB RDS

Background information

The real-time data lake solution is a next-generation data warehousing solution that features low costs and low latency. This solution supports a large number of analytics datasets on which create, read, update, and delete (CRUD) operations are performed. The following figure shows the architecture of the real-time data lake solution.

Prerequisites

• The ApsaraDB RDS instance from which you want to synchronize data is deployed in the same region as DTS and DLA.
• A data subscription task is created for DTS. This task is used to subscribe to data flows in binary log files on the ApsaraDB RDS instance that is mapped to DLA. For more information, see Overview of change tracking scenarios.
• Data flows in binary log files on the ApsaraDB RDS instance are configured for the DTS data subscription task that you created. For more information, see Track data changes from ApsaraDB RDS for MySQL (new) and Track data changes from an ApsaraDB RDS for MySQL instance (previous).
  Note To ensure the security and performance of your business, we recommend that you select the virtual private cloud (VPC) as the network type when you configure a data subscription task.

Procedure

1. Create a Spark cluster.
   1. Log on to the DLA console and create a Spark cluster. For more information, see Create a virtual cluster.
   2. Configure the VPC where the Spark cluster, ApsaraDB RDS instance, and DTS are deployed. This step ensures that you can use the serverless Spark engine of DLA to consume data in the VPC. For more information, see Access your VPC.
2. Create a full synchronization job.
   1. In the left-side navigation pane, choose Serverless Spark > Submit job.
   2. On the Parameter Configuration page, click Create Job to create a Spark job. This job is used to synchronize all data of the ApsaraDB RDS instance. For more information, see Create and run Spark jobs.
   3. After the full synchronization job is created, select the job from the Job List navigation tree and compile code in the code editor to configure the job. Sample code:

      {
          "args": [
                  "oss://<OSS directory>/dla-lakehouse-batch-jdbc-config.properties"
          ],
        "file": "oss://<OSS directory>/datalake-batch-1.0.0-shaded.jar",
        "name": "dla-lakehouse-batch", // The name of the job, which can be customized.
        "className": "com.aliyun.dla.batch.jdbc.SparkHudiJdbc",
        "conf": {
          "spark.driver.resourceSpec": "large", // The resource specifications of the Spark driver, which can be set to small, medium, large, xlarge, or 2xlarge. small indicates 1 CU. medium indicates 2 CUs. large indicates 4 CUs. xlarge indicates 8 CUs. 2xlarge indicates 16 CUs. One CU equals 1 CPU core and 4 GB of memory.
          "spark.executor.resourceSpec": "medium",  // The resource specifications of the Spark executor, which can be set to small, medium, large, xlarge, or 2xlarge. small indicates 1 CU. medium indicates 2 CUs. large indicates 4 CUs. xlarge indicates 8 CUs. 2xlarge indicates 16 CUs. One CU equals 1 CPU core and 4 GB of memory.
          "spark.executor.instances": 5,  // The number of executors. The serverless Spark engine provides only one driver.
          "spark.dla.eni.enable": "true",  // Specifies whether to use an elastic network interface (ENI) to connect to the VPC.
          "spark.dla.eni.vswitch.id": "vsw-xxx",  // The ID of the vSwitch that is deployed in the VPC.
          "spark.dla.eni.security.group.id": "sg-xxx", // The ID of the security group that is associated with the VPC.
          "spark.dla.job.log.oss.uri": "oss://<OSS directory>/spark-logs/", // The Uniform Resource Identifier (URI) of the OSS directory where Spark logs are saved.
          "spark.sql.hive.metastore.version": "dla", // The version of the Hive metastore that is automatically generated for subsequent data analysis on DLA.
          "spark.dla.connectors": "oss"    // The service from which you read and write data. In this example, the service is OSS.
        }
      }

      Note To download the datalake-batch-1.0.0-shaded.jar file, you can click this URL.

      Sample data in the dla-lakehouse-batch-jdbc-config.properties template file:

      ### JDBC configurations for the source database
      ## The username that is used to log on to the source database.
      dla.datalake.batch.jdbc.username=<JDBC_USERNAME>
      ## The password that is used to log on to the source database.
      dla.datalake.batch.jdbc.password=<JDBC_PASSWORD>
      ## The JDBC URL of the source database.
      dla.datalake.batch.jdbc.url=<JDBC_URL>
      ## The names of source databases. Multiple database names are separated by commas (,).
      dla.datalake.batch.jdbc.db.name=DBNAME
      
      ## The OSS root directory to which data is synchronized. This directory is used to build a data lakehouse.
      dla.datalake.hoodie.target.base.path=oss://lakehouse/
      
      ## The JDBC synchronization mode. Set this parameter to database.
      dla.datalake.batch.jdbc.sync.mode=database

   4. After the Spark job is configured, click Execute to run the full synchronization job.
3. Create an incremental synchronization job.
   After the full synchronization job succeeds, repeat Step 2 to create a Spark incremental synchronization job. Sample code for configuring the incremental synchronization job:

   {
       "args": [
                  "oss://<OSS directory>/dla-lakehouse-streaming-dts-config.properties"
       ],
       "file": "oss://<OSS directory>/datalake-streaming-1.0.0-shaded.jar",
       "name": "dla-lakehouse-streaming",  // The name of the job, which can be customized.
       "className": "com.aliyun.dla.streaming.dts.SparkHudiDts",
       "conf": {
             "spark.driver.resourceSpec": "large", // The resource specifications of the Spark driver, which can be set to small, medium, large, xlarge, or 2xlarge. small indicates 1 CU. medium indicates 2 CUs. large indicates 4 CUs. xlarge indicates 8 CUs. 2xlarge indicates 16 CUs. One CU equals 1 CPU core and 4 GB of memory.
               "spark.executor.resourceSpec": "medium",  // The resource specifications of the Spark executor, which can be set to small, medium, large, xlarge, or 2xlarge. small indicates 1 CU. medium indicates 2 CUs. large indicates 4 CUs. xlarge indicates 8 CUs. 2xlarge indicates 16 CUs. One CU equals 1 CPU core and 4 GB of memory.
               "spark.executor.instances": 20,  // The number of executors. The serverless Spark engine provides only one driver.
               "spark.dla.eni.enable": "true",  // Specifies whether to use an ENI to connect to the VPC.
               "spark.dla.eni.vswitch.id": "vsw-xxx",  // The ID of the vSwitch that is deployed in the VPC.
               "spark.dla.eni.security.group.id": "sg-xxx", // The ID of the security group that is associated with the VPC.
               "spark.dla.job.log.oss.uri": "oss://<OSS directory>/spark-logs/", // The URI of the OSS directory where Spark logs are saved.
               "spark.sql.hive.metastore.version": "dla", // The version of the Hive metastore that is automatically generated for subsequent data analysis on DLA.
           "spark.dla.connectors": "oss"    // The service from which you read and write data. In this example, the service is OSS.
       }
   }

   Note To download the datalake-streaming-1.0.0-shaded.jar file, you can click this URL.

   Sample data in the dla-lakehouse-streaming-dts-config.properties template file:

   ### dts config
   ## The username that is used to create a data subscription task for DTS.
   dla.datalake.streaming.dts.username=DTS_USERNAME
   ## The password that is used to create a data subscription task for DTS.
   dla.datalake.streaming.dts.password=DTS_PASSWORD
   ## The DTS consumer offset. You can set this parameter to latest, earliest, or a custom value. The default value is latest.
   dla.datalake.streaming.dts.offset=latest
   ## The start time of data consumption during an incremental synchronization job. The value is accurate to seconds. For example, 1608523200 indicates 2020/12/21 12:00:00.
   dla.datalake.streaming.dts.offset.by.timestamp=1608523200
   ## The ID of the consumer group. You can view this parameter setting in the DTS console.
   dla.datalake.streaming.dts.group.id=dtsxxxxx
   ## The endpoint of the DTS server. You can view this parameter setting in the DTS console.
   dla.datalake.streaming.dts.bootstrap.server=dts-xxx-vpc.aliyuncs.com:18003
   ## The subscription topic of DTS. You can view this topic in the DTS console.
   dla.datalake.streaming.dts.subscribe.topic=cn_hangzhou_xxx
   
   ## The storage location of checkpoints, which is used to ensure that no data is missing.
   dla.datalake.streaming.dts.checkpoint.location=oss://<OSS directory>/checkpoint/
   
   ## The OSS root directory to which data is synchronized. This directory must be the same as the directory to which full data is synchronized.
   dla.datalake.hoodie.target.base.path=oss://lakehouse/

Parameters in the dla-lakehouse-streaming-dts-config.properties file

You can specify optional parameters based on your business requirements. The required parameters must be specified in the dla-lakehouse-streaming-dts-config.properties file.