Function Compute (2.0):[Feature changes] ApsaraDB RDS triggers to be discontinued

Prerequisites

• A database account that can be used to configure DTS data subscription tasks is created. For more information, see Create an account on an ApsaraDB RDS for MySQL instance.

• EventBridge is activated and permissions are granted to a RAM user. For more information, see Activate EventBridge and grant permissions to a RAM user.

• A service and a function are created in the Function Compute console. For more information, see Create a service and Create a function.

Procedure

Configurations in the DTS console

1. Log on to the DTS console. In the top navigation bar, select a region.

2. In the left-side navigation pane, click Change Tracking. In the upper-right corner of the Change Tracking page, click Create Task.

3. On the Configure Source and Destination Databases step of the Create Task page, configure the parameters in the Source Database section and the Consumer Network Type section. Then, click Test Connectivity and Proceed.

   The following table describes the parameters that you must configure. Retain the default values for other parameters.

   Note

   Internet access is required to establish a connection for the test. Make sure that the Internet access feature is enabled for the ApsaraDB RDS instance. You can configure the Internet access feature on the Database Connection page in the ApsaraDB RDS console.

   Parameter	Description	Example
   The Source Database section
   Database Type	Select the type of the source database.	MySQL
   RDS Instance ID	Select an ApsaraDB RDS instance from which you want to track data changes.	rm-bp1pw60i18f2x****
   Database Account	The database account of the source ApsaraDB RDS instance.	db_chi
   Database Password	The password of the account of the source ApsaraDB RDS database.	*************
   The Consumer Network Type section
   VPC	Select a virtual private cloud (VPC) to which the change tracking instance belongs. Note The VPC and vSwitch that you specify in the Consumer Network Type section must be the same as the VPC and vSwitch of the ApsaraDB RDS instance. You can view the information about the VPC and vSwitch of an ApsaraDB RDS instance on the Database Connection page in the ApsaraDB RDS console.	vpc-bp12c5dzorfoizcez****
   vSwitch	Select the vSwitch to which the change tracking instance belongs. Note The VPC and vSwitch that you specify in the Consumer Network Type section must be the same as the VPC and vSwitch of the ApsaraDB RDS instance. You can view the information about the VPC and vSwitch of an ApsaraDB RDS instance on the Database Connection page in the ApsaraDB RDS console.	vsw-bp1lt2oxenx87jvc8****

4. In the Configure Objects and Advanced Settings step of the Create Task page, select a data type and a table. Then, follow the on-screen instructions to complete the configurations in the Advanced Settings step and the Precheck step.

5. In the Purchase Instance step of the Create Task page, follow the on-screen instructions to purchase the instance. Then, start the task in the task list.

6. Click the ID of the task that you created. On the Task Management page, click Consume Data. On the page that appears, click Add Consumer Group. In the Add Consumer Group panel, configure the following parameters: Consumer Group Name, Account, and Password.

Configurations in the EventBridge console

1. Log on to the EventBridge console. In the left-side navigation pane, click Event Streams.
2. In the top navigation bar, select a region and click Create Event Stream.

3. In the Create EventStreaming panel, configure the parameters and click OK.

   1. In the Basic Information step, configure the EventStreaming Name parameter and the Description parameter, and click Next Step.

   2. In the Event Source step, select Data Transmission Service (DTS) from the Event Provider drop-down list, select the consumer group that you specified for the Consumer Group Name parameter in step 6 of the preceding section in this topic from the Consumer Group drop-down list, and then configure the Password parameter and the Consumer Offset parameter. Click Next Step.

   3. Optional: In the Rule step, configure the event rule and click Next Step.

   4. In the Event Target step, set the Service Type parameter to Function Compute and configure the Service parameter and the Function parameter.

   After the event stream is created, you can view the event stream whose Event Source is DTS and Event Target is Function Compute on the Event Streams page. The event stream is in the Running state.

After you complete the preceding configurations, the new path ApsaraDB RDS > DTS > EventBridge > Function Compute is configured.

Modify function code based on the event format

The following section provides demos for event parsing by using the original path and the new path in Python and Java. You can view the differences between the paths and modify the event parsing logic of the function.

Python

• Sample code for event parsing that uses the ApsaraDB RDS > Function Compute path

  # -*- coding: utf-8 -*-
  import json
  import logging
  
  def handler(event, context):
    logger = logging.getLogger()
    eventObj = json.loads(event)
    logger.info("rds trigger event = {}".format(eventObj))
    # You can add your processing logic in the code. Example: Update Redis cache. 
    return "OK"

• Sample code for event parsing that uses the ApsaraDB RDS > DTS > EventBridge > Function Compute path

  # -*- coding: utf-8 -*-
  import logging
  import json
  
  def handler(event, context):
    logger = logging.getLogger()
    # Convert the input parameter of the event to a string list. 
    evt_list = json.loads(event)
    for evt in evt_list:
        # Convert each element in the string list to data of the JSON type. 
        evt_obj = json.loads(evt)
        logger.info(evt_obj)
    return "OK"

Java

• Sample code for event parsing that uses the ApsaraDB RDS > Function Compute path

  package example;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  
  import com.google.gson.*;
  import com.google.gson.JsonParser;
  import com.google.gson.JsonObject;
  import com.aliyun.fc.runtime.Context;
  import com.aliyun.fc.runtime.StreamRequestHandler;
  
  import java.io.*;
  import java.nio.charset.StandardCharsets;
  
  public class App implements StreamRequestHandler {
      @Override
      public void handleRequest(
              InputStream inputStream, OutputStream outputStream, Context context) throws IOException {
            // Read data from the input parameters and store the data in ByteBuffer. 
          ByteArrayOutputStream result = new ByteArrayOutputStream();
          byte[] buffer = new byte[1024];
          for (int length; (length = inputStream.read(buffer)) != -1; ) {
              result.write(buffer, 0, length);
          }
  
          // Convert data of the byte type to data of the object type. 
          String stringData = result.toString(StandardCharsets.UTF_8.name());
          JsonObject jsonObj = (JsonObject)(new JsonParser().parse(stringData));
          System.out.println(jsonObj);
          outputStream.write(new String("OK").getBytes());
      }
  }
                      

  The following sample code provides an example on the pom.xml file:

  <project xmlns="http://maven.apache.org/POM/4.0.0"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>example</groupId>
    <artifactId>FCJavaDemo</artifactId>
    <packaging>jar</packaging>
    <version>1.0-SNAPSHOT</version>
    <name>FCJavaDemo</name>
  
    <dependencies>
      <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>3.8.1</version>
        <scope>test</scope>
      </dependency>
      <dependency>
        <groupId>com.aliyun.fc.runtime</groupId>
        <artifactId>fc-java-core</artifactId>
        <version>1.4.1</version>
      </dependency>
      <dependency>
        <groupId>com.google.code.gson</groupId>
        <artifactId>gson</artifactId>
        <version>2.8.5</version>
      </dependency>
      <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
      </dependency>
    </dependencies>
  
    <build>
      <plugins>
        <plugin>
          <artifactId>maven-assembly-plugin</artifactId>
          <configuration>
            <descriptorRefs>
              <descriptorRef>jar-with-dependencies</descriptorRef>
            </descriptorRefs>
          </configuration>
          <executions>
            <execution>
              <id>make-my-jar-with-dependencies</id>
              <phase>package</phase>
              <goals>
                <goal>single</goal>
              </goals>
            </execution>
          </executions>
        </plugin>
      </plugins>
    </build>
  
    <properties>
      <maven.compiler.target>1.8</maven.compiler.target>
      <maven.compiler.source>1.8</maven.compiler.source>
      <maven.test.skip>true</maven.test.skip>
    </properties>
  </project>

• Sample code for event parsing that uses the ApsaraDB RDS > DTS > EventBridge > Function Compute path

  package example;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  
  import com.google.gson.*;
  import com.google.gson.JsonParser;
  import com.google.gson.JsonObject;
  import com.aliyun.fc.runtime.Context;
  import com.aliyun.fc.runtime.StreamRequestHandler;
  
  import java.io.*;
  import java.nio.charset.StandardCharsets;
  
  public class App implements StreamRequestHandler {
  
      @Override
      public void handleRequest(InputStream inputStream, OutputStream outputStream, Context context) throws IOException {
            // Read data from the input parameters and store the data in ByteBuffer.
          ByteArrayOutputStream result = new ByteArrayOutputStream();
          byte[] buffer = new byte[1024];
          for (int length; (length = inputStream.read(buffer)) != -1; ) {
              result.write(buffer, 0, length);
          }
  
      // Convert data of the byte type to data of the string type. 
          String stringData = result.toString(StandardCharsets.UTF_8.name());
          Gson gson =new Gson();
          String[] stringArrayData = gson.fromJson(stringData, String[].class);
  
          // Scan each element in the string and convert each element into data of the JSON type. The data field is the original data of DTS. 
          for(String elem : stringArrayData){
              JsonObject jsonObj = (JsonObject)(new JsonParser().parse(elem));
              System.out.println(jsonObj.get("data"));
          }
          outputStream.write((new String("OK")).getBytes());
      }
  }

  The following sample code provides an example of the pom.xml file:

  <project xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>example</groupId>
    <artifactId>FCJavaDemo</artifactId>
    <packaging>jar</packaging>
    <version>1.0-SNAPSHOT</version>
    <name>FCJavaDemo</name>
  
    <dependencies>
      <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>3.8.1</version>
        <scope>test</scope>
      </dependency>
      <dependency>
        <groupId>com.aliyun.fc.runtime</groupId>
        <artifactId>fc-java-core</artifactId>
        <version>1.4.1</version>
      </dependency>
      <dependency>
        <groupId>com.google.code.gson</groupId>
        <artifactId>gson</artifactId>
        <version>2.8.5</version>
      </dependency>
      <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
      </dependency>
    </dependencies>
  
    <build>
      <plugins>
        <plugin>
          <artifactId>maven-assembly-plugin</artifactId>
          <configuration>
            <descriptorRefs>
              <descriptorRef>jar-with-dependencies</descriptorRef>
            </descriptorRefs>
          </configuration>
          <executions>
            <execution>
              <id>make-my-jar-with-dependencies</id>
              <phase>package</phase>
              <goals>
                <goal>single</goal>
              </goals>
            </execution>
          </executions>
        </plugin>
      </plugins>
    </build>
  
    <properties>
      <maven.compiler.target>1.8</maven.compiler.target>
      <maven.compiler.source>1.8</maven.compiler.source>
      <maven.test.skip>true</maven.test.skip>
    </properties>
  </project>