Real-time transaction data analytics with Flink

Step 1: Create data tables and a tunnel

Use Tablestore SDK for Java to create the source table (source_order) and the sink table (sink_order), and create an incremental tunnel on the source table for Flink to consume.

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.TunnelClient;
import com.alicloud.openservices.tablestore.core.ResourceManager;
import com.alicloud.openservices.tablestore.core.auth.*;
import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.model.tunnel.*;

import java.util.*;

public class CreateTablesAndTunnel {
    public static void main(String[] args) {
        // Initialize V4 signature client.
        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 v4Credentials = V4Credentials.createByServiceCredentials(credentials, region);
        CredentialsProvider provider = new DefaultCredentialProvider(v4Credentials);
        SyncClient client = new SyncClient(endpoint, provider, instanceName, null, new ResourceManager(null, null));

        // 1. Create source_order table (PK: metering + orderid).
        createTable(client, "source_order", Arrays.asList(
            new PrimaryKeySchema("metering", PrimaryKeyType.STRING),
            new PrimaryKeySchema("orderid", PrimaryKeyType.STRING)));

        // 2. Create sink_order table (PK: metering + ts).
        createTable(client, "sink_order", Arrays.asList(
            new PrimaryKeySchema("metering", PrimaryKeyType.STRING),
            new PrimaryKeySchema("ts", PrimaryKeyType.INTEGER)));

        // 3. Create stream tunnel on source_order for Flink consumption.
        TunnelClient tunnelClient = new TunnelClient(endpoint, accessKeyId, accessKeySecret, instanceName);
        CreateTunnelRequest tunnelRequest = new CreateTunnelRequest(
            "source_order", "flink_agg", TunnelType.Stream);
        CreateTunnelResponse tunnelResponse = tunnelClient.createTunnel(tunnelRequest);
        System.out.println("Tunnel flink_agg created.");

        tunnelClient.shutdown();
        client.shutdown();
    }

    private static void createTable(SyncClient client, String tableName, List<PrimaryKeySchema> pkSchema) {
        TableMeta meta = new TableMeta(tableName);
        for (PrimaryKeySchema pk : pkSchema) {
            meta.addPrimaryKeyColumn(pk);
        }
        TableOptions options = new TableOptions(-1, 1);
        CreateTableRequest request = new CreateTableRequest(meta, options);
        try {
            client.createTable(request);
            System.out.println("Table " + tableName + " created.");
        } catch (Exception e) {
            if (e.getMessage() != null && e.getMessage().contains("already exist")) {
                System.out.println("Table " + tableName + " already exists.");
            } else {
                throw new RuntimeException(e);
            }
        }
    }
}

Step 2: Create and start a Flink stream processing job

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.

   The Flink SQL defines temporary tables for the source table and the sink table using the Tablestore ots connector. The source table specifies the tunnel name through the tunnelName parameter. Flink reads incremental data from this tunnel. The sink table specifies the attribute columns to write through the valueColumns parameter. The INSERT INTO ... SELECT statement starts stream processing, aggregating the order count and total order amount in 1-minute tumbling windows.

   -- Create source table: read incremental data from Tablestore via Tunnel.
   CREATE TEMPORARY TABLE tablestore_input (
       metering VARCHAR,
       orderid VARCHAR,
       price DOUBLE,
       byerid BIGINT,
       sellerid BIGINT,
       productid BIGINT,
       ts BIGINT,
       ptime AS TO_TIMESTAMP(ts * 1000),
       WATERMARK FOR ptime AS ptime - INTERVAL '2' SECOND
   ) WITH (
       'connector' = 'ots',
       'endPoint' = '<endpoint>',
       'instanceName' = '<instance-name>',
       'tableName' = 'source_order',
       'tunnelName' = 'flink_agg',
       'accessId' = '<AccessKey ID>',
       'accessKey' = '<AccessKey Secret>'
   );
   
   -- Create sink table: write aggregation results to Tablestore.
   CREATE TEMPORARY TABLE tablestore_output (
       metering VARCHAR,
       ts BIGINT,
       price DOUBLE,
       ordercount BIGINT,
       PRIMARY KEY(metering, ts) NOT ENFORCED
   ) WITH (
       'connector' = 'ots',
       'endPoint' = '<endpoint>',
       'instanceName' = '<instance-name>',
       'tableName' = 'sink_order',
       'accessId' = '<AccessKey ID>',
       'accessKey' = '<AccessKey Secret>',
       'valueColumns' = 'price,ordercount'
   );
   
   -- 1-minute tumbling window aggregation.
   INSERT INTO tablestore_output
   SELECT metering,
          UNIX_TIMESTAMP(CAST(TUMBLE_START(ptime, INTERVAL '1' MINUTE) AS STRING)) AS ts,
          SUM(price) AS price,
          COUNT(orderid) AS ordercount
   FROM tablestore_input
   GROUP BY TUMBLE(ptime, INTERVAL '1' MINUTE), metering;

   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.
   tunnelName	The name of the tunnel for the source table. Only the source table requires this parameter. Flink reads incremental data from this tunnel.
   accessId / accessKey	The AccessKey ID and AccessKey Secret of your Alibaba Cloud account.
   valueColumns	The attribute columns to write to the sink table, separated by commas. Only the sink table requires this parameter.

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 aggregation results

The following code simulates real-time transactions by writing batches of 10 orders every 5 seconds to source_order and polling sink_order every 30 seconds to display aggregation results per 1-minute window.

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

import java.util.*;

public class RealtimeTradeDemo {
    public static void main(String[] args) throws InterruptedException {
        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 v4Credentials = V4Credentials.createByServiceCredentials(credentials, region);
        CredentialsProvider provider = new DefaultCredentialProvider(v4Credentials);
        SyncClient client = new SyncClient(endpoint, provider, instanceName, null, new ResourceManager(null, null));

        Random random = new Random();
        String metering = "web";
        int totalOrders = 0;

        // Write data and verify results for about 3 minutes.
        for (int round = 0; round < 36; round++) {
            // Write 10 orders per batch.
            long now = System.currentTimeMillis() / 1000;
            BatchWriteRowRequest writeRequest = new BatchWriteRowRequest();
            for (int i = 0; i < 10; i++) {
                String orderId = String.format("ord_%d_%04d", now, i);
                double price = Math.round((10.0 + random.nextDouble() * 990.0) * 100.0) / 100.0;
                PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                    .addPrimaryKeyColumn("metering", PrimaryKeyValue.fromString(metering))
                    .addPrimaryKeyColumn("orderid", PrimaryKeyValue.fromString(orderId))
                    .build();
                RowPutChange row = new RowPutChange("source_order", pk);
                row.addColumn("price", ColumnValue.fromDouble(price));
                row.addColumn("byerid", ColumnValue.fromLong(1000 + random.nextInt(100)));
                row.addColumn("sellerid", ColumnValue.fromLong(2000 + random.nextInt(50)));
                row.addColumn("productid", ColumnValue.fromLong(3000 + random.nextInt(200)));
                row.addColumn("ts", ColumnValue.fromLong(now + i));
                writeRequest.addRowChange(row);
            }
            client.batchWriteRow(writeRequest);
            totalOrders += 10;

            // Every 30 seconds (6 rounds), read and display aggregation results.
            if ((round + 1) % 6 == 0) {
                System.out.println("\n--- Aggregation results (total " + totalOrders + " orders written) ---");
                RangeRowQueryCriteria criteria = new RangeRowQueryCriteria("sink_order");
                criteria.setInclusiveStartPrimaryKey(PrimaryKeyBuilder.createPrimaryKeyBuilder()
                    .addPrimaryKeyColumn("metering", PrimaryKeyValue.INF_MIN)
                    .addPrimaryKeyColumn("ts", PrimaryKeyValue.INF_MIN).build());
                criteria.setExclusiveEndPrimaryKey(PrimaryKeyBuilder.createPrimaryKeyBuilder()
                    .addPrimaryKeyColumn("metering", PrimaryKeyValue.INF_MAX)
                    .addPrimaryKeyColumn("ts", PrimaryKeyValue.INF_MAX).build());
                criteria.setMaxVersions(1);
                criteria.setLimit(20);
                GetRangeResponse response = client.getRange(new GetRangeRequest(criteria));
                for (Row row : response.getRows()) {
                    long ts = row.getPrimaryKey().getPrimaryKeyColumn("ts").getValue().asLong();
                    double totalPrice = row.getColumn("price").get(0).getValue().asDouble();
                    long orderCount = row.getColumn("ordercount").get(0).getValue().asLong();
                    System.out.println(String.format("  Window %s: orders=%d, totalPrice=%.2f",
                        new java.text.SimpleDateFormat("HH:mm:ss").format(new java.util.Date(ts * 1000)),
                        orderCount, totalPrice));
                }
            }
            Thread.sleep(5000);
        }
        client.shutdown();
    }
}

Sample output:

--- Aggregation results (total 60 orders written) ---
  Window 14:17:00: orders=103, totalPrice=48723.75
  Window 14:18:00: orders=118, totalPrice=62093.70

Each line shows the aggregation result for one 1-minute window. As new transactions arrive, new windows appear and existing window totals update accordingly.

Step 4: (Optional) Visualize results with DataV

To display aggregation results on a dashboard, use the DataV data visualization service with a Tablestore data source.

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

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