Real-time correlation analysis of transaction data with Flink

Step 1: Create tables and write product data

Create the transaction record source table in RDS MySQL.

CREATE TABLE consume_record (
    consume_id varchar(20) NOT NULL,
    product_id varchar(20) NOT NULL,
    consume_time bigint(20) NOT NULL,
    consume_name varchar(20) NOT NULL,
    consume_phone varchar(20) NOT NULL,
    PRIMARY KEY (consume_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

Use the Tablestore SDK for Java to create the product information table and result table, and write product data to the product information table.

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.core.ResourceManager;
import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.core.auth.DefaultCredentials;
import com.alicloud.openservices.tablestore.core.auth.V4Credentials;
import com.alicloud.openservices.tablestore.core.auth.DefaultCredentialProvider;

public class FlinkBigdataSetup {
    public static void main(String[] args) {
        String endpoint = System.getenv("OTS_ENDPOINT");
        String accessKeyId = System.getenv("OTS_AK_ENV");
        String accessKeySecret = System.getenv("OTS_SK_ENV");
        String instanceName = System.getenv("OTS_INSTANCE");
        String region = System.getenv("OTS_REGION");

        DefaultCredentials credentials = new DefaultCredentials(accessKeyId, accessKeySecret);
        V4Credentials v4 = V4Credentials.createByServiceCredentials(credentials, region);
        DefaultCredentialProvider provider = new DefaultCredentialProvider(v4);
        SyncClient client = new SyncClient(endpoint, provider, instanceName, null,
                new ResourceManager(null, null));

        // Create the product table (primary key: product_id)
        TableMeta productMeta = new TableMeta("product");
        productMeta.addPrimaryKeyColumn("product_id", PrimaryKeyType.STRING);
        TableOptions options = new TableOptions(-1, 1);
        CreateTableRequest createReq = new CreateTableRequest(productMeta, options);
        createReq.setReservedThroughput(new ReservedThroughput(0, 0));
        client.createTable(createReq);

        // Create the consumption result table (primary key: consume_id + product_id)
        TableMeta resultMeta = new TableMeta("consume_product");
        resultMeta.addPrimaryKeyColumn("consume_id", PrimaryKeyType.STRING);
        resultMeta.addPrimaryKeyColumn("product_id", PrimaryKeyType.STRING);
        createReq = new CreateTableRequest(resultMeta, options);
        createReq.setReservedThroughput(new ReservedThroughput(0, 0));
        client.createTable(createReq);

        // Insert product data
        String[][] products = {
            {"P001", "15.5", "Food"},       {"P002", "89.0", "Clothing"},
            {"P003", "2999.0", "Electronics"}, {"P004", "45.0", "Daily necessities"},
            {"P005", "128.0", "Beauty"},     {"P006", "35.0", "Beverages"},
            {"P007", "599.0", "Sports"},     {"P008", "12.0", "Stationery"},
            {"P009", "268.0", "Home decor"}, {"P010", "1599.0", "Home appliances"}
        };
        for (String[] p : products) {
            PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                    .addPrimaryKeyColumn("product_id", PrimaryKeyValue.fromString(p[0]))
                    .build();
            RowPutChange change = new RowPutChange("product", pk);
            change.addColumn("price", ColumnValue.fromDouble(Double.parseDouble(p[1])));
            change.addColumn("product_type", ColumnValue.fromString(p[2]));
            client.putRow(new PutRowRequest(change));
        }
        System.out.println("Tables created and " + products.length + " product records inserted.");
        client.shutdown();
    }
}

Step 2: Create and start a Flink real-time correlation job

Create a Flink SQL streaming job in the Realtime Compute for Apache Flink console. The job uses the MySQL CDC connector to capture transaction record changes in real time, joins them with Tablestore product information, and writes the results to the result table.

1. Log on to the Realtime Compute console and click the instance name to open the workspace.

2. Click Development > ETL, and then click + > New Blank Stream Draft.

3. In the dialog box, enter a Name and click Create.

4. Paste the following Flink SQL code in the editor.Replace the parameter values with your actual values.

   -- Data source: RDS MySQL CDC
   CREATE TEMPORARY TABLE mysql_source (
       consume_id   VARCHAR,
       product_id   VARCHAR,
       consume_time BIGINT,
       consume_name VARCHAR,
       consume_phone VARCHAR,
       PRIMARY KEY (consume_id) NOT ENFORCED
   ) WITH (
       'connector' = 'mysql-cdc',
       'hostname' = '<RDS internal endpoint>',
       'port' = '3306',
       'database-name' = '<database name>',
       'table-name' = 'consume_record',
       'username' = '<username>',
       'password' = '<password>'
   );
   
   -- Dimension table: Tablestore product table
   CREATE TEMPORARY TABLE ots_product (
       product_id   VARCHAR,
       price        DOUBLE,
       product_type VARCHAR,
       PRIMARY KEY (product_id) NOT ENFORCED
   ) WITH (
       'connector' = 'ots',
       'endPoint' = 'https://<instance name>.<region>.vpc.tablestore.aliyuncs.com',
       'instanceName' = '<instance name>',
       'tableName' = 'product',
       'accessId' = '<AccessKey ID>',
       'accessKey' = '<AccessKey Secret>'
   );
   
   -- Sink table: Tablestore result table
   CREATE TEMPORARY TABLE ots_sink (
       consume_id    VARCHAR,
       product_id    VARCHAR,
       price         DOUBLE,
       consume_time  BIGINT,
       consume_name  VARCHAR,
       consume_phone VARCHAR,
       product_type  VARCHAR,
       PRIMARY KEY (consume_id, product_id) NOT ENFORCED
   ) WITH (
       'connector' = 'ots',
       'endPoint' = 'https://<instance name>.<region>.vpc.tablestore.aliyuncs.com',
       'instanceName' = '<instance name>',
       'tableName' = 'consume_product',
       'accessId' = '<AccessKey ID>',
       'accessKey' = '<AccessKey Secret>',
       'valueColumns' = 'price,consume_time,consume_name,consume_phone,product_type'
   );
   
   -- Join query: associate each consumption record with product information
   INSERT INTO ots_sink
   SELECT
       s.consume_id,
       s.product_id,
       p.price,
       s.consume_time,
       s.consume_name,
       s.consume_phone,
       p.product_type
   FROM mysql_source AS s
   JOIN ots_product FOR SYSTEM_TIME AS OF PROCTIME() AS p
       ON s.product_id = p.product_id;

   The following tables describe the connector parameters in the Flink SQL code.

   mysql-cdc connector parameters

   Parameter	Description
   connector	The connector type. Set to mysql-cdc.
   hostname	The internal endpoint of the RDS MySQL instance.
   port	The port number of the RDS MySQL instance. Default value: 3306.
   database-name	The name of the RDS MySQL database.
   table-name	The name of the MySQL source table.
   username	The account for the RDS MySQL database.
   password	The password for the RDS MySQL database.

   ots connector parameters

   Parameter	Description
   connector	The connector type. Set to ots. For more information, see Tablestore connector.
   endPoint	The VPC endpoint of the Tablestore instance.
   instanceName	The name of the Tablestore instance.
   tableName	The name of the Tablestore data table.
   accessId	The AccessKey ID of your Alibaba Cloud account.
   accessKey	The AccessKey secret of your Alibaba Cloud account.
   valueColumns	The attribute column names to write to the result table. Separate multiple names with commas (,). Required only for sink tables.

   Note

   • The MySQL CDC connector reads the full snapshot of the source table when the job starts, and then automatically switches to incremental mode to continuously capture data changes.

   • Flink uses the FOR SYSTEM_TIME AS OF PROCTIME() syntax to query the latest product information from Tablestore in real time as each transaction record arrives.

5. Click Deploy > Confirm in the upper-right corner, and then click Operations > Start to start the job.

Step 3: Write test data and verify results

After the Flink job starts, write transaction records to the MySQL source table. Flink captures the changes, joins them with product information, and writes the results to the consume_product table in Tablestore. The following code writes test data and reads the joined results for verification.

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.core.ResourceManager;
import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.core.auth.DefaultCredentials;
import com.alicloud.openservices.tablestore.core.auth.V4Credentials;
import com.alicloud.openservices.tablestore.core.auth.DefaultCredentialProvider;

import java.sql.*;
import java.util.Random;

public class FlinkBigdataVerify {
    public static void main(String[] args) throws Exception {
        // Write test data to MySQL
        String mysqlUrl = String.format("jdbc:mysql://%s:%s/%s?useSSL=false",
                System.getenv("MYSQL_HOST"), System.getenv("MYSQL_PORT"), System.getenv("MYSQL_DB"));
        Connection conn = DriverManager.getConnection(mysqlUrl,
                System.getenv("MYSQL_USER"), System.getenv("MYSQL_PASS"));

        String[] productIds = {"P001","P002","P003","P004","P005","P006","P007","P008","P009","P010"};
        String[] names = {"Alice","Bob","Charlie","Diana","Eve"};
        Random rand = new Random();
        long now = System.currentTimeMillis() / 1000;

        PreparedStatement ps = conn.prepareStatement(
                "INSERT INTO consume_record VALUES (?,?,?,?,?)");
        for (int i = 1; i <= 20; i++) {
            ps.setString(1, String.format("C%04d", i));
            ps.setString(2, productIds[rand.nextInt(productIds.length)]);
            ps.setLong(3, now + i);
            ps.setString(4, names[rand.nextInt(names.length)]);
            ps.setString(5, String.format("138%08d", rand.nextInt(100000000)));
            ps.executeUpdate();
        }
        System.out.println("Inserted 20 test records to MySQL.");
        conn.close();

        // Wait for Flink to process
        System.out.println("Waiting 60 seconds for Flink to process...");
        Thread.sleep(60000);

        // Verify result table in Tablestore
        String endpoint = System.getenv("OTS_ENDPOINT");
        String ak = System.getenv("OTS_AK_ENV");
        String sk = System.getenv("OTS_SK_ENV");
        String instance = System.getenv("OTS_INSTANCE");
        String region = System.getenv("OTS_REGION");

        DefaultCredentials cred = new DefaultCredentials(ak, sk);
        V4Credentials v4 = V4Credentials.createByServiceCredentials(cred, region);
        DefaultCredentialProvider provider = new DefaultCredentialProvider(v4);
        SyncClient client = new SyncClient(endpoint, provider, instance, null,
                new ResourceManager(null, null));

        RangeRowQueryCriteria criteria = new RangeRowQueryCriteria("consume_product");
        criteria.setInclusiveStartPrimaryKey(PrimaryKeyBuilder.createPrimaryKeyBuilder()
                .addPrimaryKeyColumn("consume_id", PrimaryKeyValue.INF_MIN)
                .addPrimaryKeyColumn("product_id", PrimaryKeyValue.INF_MIN).build());
        criteria.setExclusiveEndPrimaryKey(PrimaryKeyBuilder.createPrimaryKeyBuilder()
                .addPrimaryKeyColumn("consume_id", PrimaryKeyValue.INF_MAX)
                .addPrimaryKeyColumn("product_id", PrimaryKeyValue.INF_MAX).build());
        criteria.setMaxVersions(1);

        GetRangeResponse resp = client.getRange(new GetRangeRequest(criteria));
        int count = 0;
        for (Row row : resp.getRows()) {
            count++;
            System.out.printf("%s | %s | price=%.1f | type=%s | name=%s%n",
                    row.getPrimaryKey().getPrimaryKeyColumn("consume_id").getValue(),
                    row.getPrimaryKey().getPrimaryKeyColumn("product_id").getValue(),
                    row.getLatestColumn("price").getValue().asDouble(),
                    row.getLatestColumn("product_type").getValue().asString(),
                    row.getLatestColumn("consume_name").getValue().asString());
        }
        System.out.println("Total: " + count + " rows in result table.");
        client.shutdown();
    }
}

Sample output:

C0001 | P007 | price=599.0 | type=Sports | name=Alice
C0002 | P003 | price=2999.0 | type=Electronics | name=Bob
C0003 | P001 | price=15.5 | type=Food | name=Charlie
...
Total: 20 rows in result table.

Step 4: (Optional) Visualize results with DataV

To visualize the joined results, connect DataV to the Tablestore result table and aggregate transaction amounts by product category.

• To activate DataV, see Activate DataV-Board Service.

• To add Tablestore as a data source, see Add a Tablestore data source.