Tablestore:Getting started

Usage notes

• By default, the system starts a thread pool to read and process data based on TunnelWorkerConfig. If you want to start multiple TunnelWorkers on a single server, we recommend that you use the same TunnelWorkerConfig to configure all TunnelWorkers.

• The incremental logs for a tunnel are retained for a maximum of seven days. The specific storage duration of incremental logs is consistent with that of the stream logs for a data table. If you create a tunnel to consume differential or incremental data, take note of the following items:

  • During full data consumption, if the tunnel does not complete the consumption of full data within seven days, an OTSTunnelExpired error occurs when the tunnel starts to consume incremental data. As a result, the tunnel cannot consume incremental data. If you estimate that full data cannot be consumed within seven days, contact Tablestore technical support.

  • During incremental data consumption, if the tunnel has not consumed incremental data for more than seven days, the consumption starts from the latest data that can be consumed. In this case, some data may not be consumed.

    Note

    If incremental data has not been consumed for more than seven days, the incremental data expires. The specific storage duration of incremental logs is consistent with that of the stream logs for a data table. If the incremental data is expired for the specified period, which is seven days by default, Tablestore disables the tunnel. In this case, the tunnel cannot be used to consume data.

• TunnelWorker requires time to warm up for initialization, which is determined by the heartbeatIntervalInSec parameter in TunnelWorkerConfig. You can use the setHeartbeatIntervalInSec method in TunnelWorkerConfig to configure this parameter. Default value: 30. Minimum value: 5. Unit: seconds.

• When the mode switches from the full channel to the incremental channel, the full channel is closed and the incremental channel is started. This process requires another period of time for initialization, which is also determined by the heartbeatIntervalInSec parameter.

• When the TunnelWorker client is shut down due to an unexpected exit or manual termination, TunnelWorker uses one of the following methods to automatically recycle resources: release the thread pool, automatically call the shutdown method that you have registered for the Channel class, or shut down the tunnel.

Prerequisites

• The following operations are performed in the Resource Access Management (RAM) console:

  • A RAM user is created and the AliyunOTSFullAccess policy is attached to the RAM user to grant the RAM user the permissions to manage Tablestore. For more information, see Create a RAM user and Grant permissions to a RAM user.

    Note

    In an actual business environment, we recommend that you create a custom policy to grant only the required permissions to the RAM user based on the principle of least privilege. This prevents security risks caused by excessive user permissions.

  • An AccessKey pair is created for the RAM user. For more information, see Create an AccessKey pair.

    Warning

    If the AccessKey pair of your Alibaba Cloud account is leaked, all resources within the account are exposed to potential risks. We recommend that you use the AccessKey pair of a RAM user to perform operations. This prevents leakage of the AccessKey pair of your Alibaba Cloud account.

• The following operations are performed in the Tablestore console:

  • A data table is created. For more information, see Use the Tablestore console, Use the Tablestore CLI, and Use Tablestore SDKs.

  • The endpoint of the instance in which the data table resides is obtained. For more information, see Obtain an endpoint of a Tablestore instance.

  • Access credentials are configured. For more information, see Configure access credentials.

Use Tunnel Service

Use Tablestore SDK for Java to get started with Tunnel Service.

1. Initialize a TunnelClient instance.

   Note

   Make sure that the TABLESTORE_ACCESS_KEY_ID and TABLESTORE_ACCESS_KEY_SECRET environment variables are configured. The TABLESTORE_ACCESS_KEY_ID environment variable specifies the AccessKey ID of your Alibaba Cloud account or RAM user. The TABLESTORE_ACCESS_KEY_SECRET environment variable specifies the AccessKey secret of your Alibaba Cloud account or RAM user.

   // Set the endPoint parameter to the endpoint of the Tablestore instance. Example: https://instance.cn-hangzhou.ots.aliyuncs.com. 
   // Set the accessKeyId parameter to the AccessKey ID and the accessKeySecret parameter to the AccessKey secret that you want to use to access Tablestore. 
   // Set the instanceName parameter to the name of the instance. 
   final String endPoint = "";
   final String accessKeyId = System.getenv("TABLESTORE_ACCESS_KEY_ID");
   final String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
   final String instanceName = "";
   TunnelClient tunnelClient = new TunnelClient(endPoint, accessKeyId, accessKeySecret, instanceName);

2. Create a tunnel.

   Before you create a tunnel, create a data table for testing or prepare an existing table. You can create a table in the Tablestore console or by using the createTable method of a SyncClient.

   // You can create three types of tunnels: TunnelType.BaseData, TunnelType.Stream, and TunnelType.BaseAndStream. 
   // The following code provides an example of how to create a BaseAndStream tunnel. To create a tunnel of a different type, set TunnelType in CreateTunnelRequest to the desired type. 
   final String tableName = "testTable";
   final String tunnelName = "testTunnel";
   CreateTunnelRequest request = new CreateTunnelRequest(tableName, tunnelName, TunnelType.BaseAndStream);
   CreateTunnelResponse resp = tunnelClient.createTunnel(request);
   // Use the tunnelId parameter to initialize a TunnelWorker. You can call the ListTunnel or DescribeTunnel operation to obtain the tunnel ID. 
   String tunnelId = resp.getTunnelId(); 
   System.out.println("Create Tunnel, Id: " + tunnelId);

3. Specify a custom data consumption callback function to start automatic data consumption. The following table describes the configurations of the TunnelClient instance.

   // Specify the callback function for data consumption to call the IChannelProcessor operation, which specifies the process and shutdown methods. 
   private static class SimpleProcessor implements IChannelProcessor {
       @Override
       public void process(ProcessRecordsInput input) {
           // The ProcessRecordsInput parameter contains the data that you obtain. 
           System.out.println("Default record processor, would print records count");
           System.out.println(
               // The NextToken parameter is used by the tunnel client to paginate data. 
               String.format("Process %d records, NextToken: %s", input.getRecords().size(), input.getNextToken()));
           try {
               // Simulate data consumption and processing. 
               Thread.sleep(1000);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
       @Override
       public void shutdown() {
           System.out.println("Mock shutdown");
       }
   }
   
   // By default, the system starts a thread pool to read and process data based on TunnelWorkerConfig. If you use a single server, multiple TunnelWorkers are started. 
   // We recommend that you configure TunnelWorkers by using the same TunnelWorkerConfig. TunnelWorkerConfig provides more advanced parameters. 
   TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
   // Configure TunnelWorkers and start automatic data processing. 
   TunnelWorker worker = new TunnelWorker(tunnelId, tunnelClient, config);
   try {
       worker.connectAndWorking();
   } catch (Exception e) {
       e.printStackTrace();
       config.shutdown();
       worker.shutdown();
       tunnelClient.shutdown();
   }

Configure TunnelWorkerConfig

TunnelWorkerConfig allows you to customize parameters for a TunnelClient instance based on your requirements. The following table describes the parameters.

Appendix: Sample code

import com.alicloud.openservices.tablestore.TunnelClient;
import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelRequest;
import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelResponse;
import com.alicloud.openservices.tablestore.model.tunnel.TunnelType;
import com.alicloud.openservices.tablestore.tunnel.worker.IChannelProcessor;
import com.alicloud.openservices.tablestore.tunnel.worker.ProcessRecordsInput;
import com.alicloud.openservices.tablestore.tunnel.worker.TunnelWorker;
import com.alicloud.openservices.tablestore.tunnel.worker.TunnelWorkerConfig;
public class TunnelQuickStart {
    private static class SimpleProcessor implements IChannelProcessor {
        @Override
        public void process(ProcessRecordsInput input) {
            System.out.println("Default record processor, would print records count");
            System.out.println(
                // The NextToken parameter is used by the tunnel client to paginate data. 
                String.format("Process %d records, NextToken: %s", input.getRecords().size(), input.getNextToken()));
            try {
                // Simulate data consumption and processing. 
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        @Override
        public void shutdown() {
            System.out.println("Mock shutdown");
        }
    }
    public static void main(String[] args) throws Exception {
        //1. Initialize the tunnel client. 
        final String endPoint = "";
        final String accessKeyId = System.getenv("OTS_AK_ENV");
        final String accessKeySecret = System.getenv("OTS_SK_ENV");
        final String instanceName = "";
        TunnelClient tunnelClient = new TunnelClient(endPoint, accessKeyId, accessKeySecret, instanceName);
        //2. Create a tunnel. Before you perform this step, you must create a table for testing. You can create a table in the Tablestore console or by using the createTable method of a SyncClient. 
        final String tableName = "testTable";
        final String tunnelName = "testTunnel";
        CreateTunnelRequest request = new CreateTunnelRequest(tableName, tunnelName, TunnelType.BaseAndStream);
        CreateTunnelResponse resp = tunnelClient.createTunnel(request);
        // Use the tunnelId parameter to initialize a TunnelWorker. You can call the ListTunnel or DescribeTunnel operation to obtain the tunnel ID. 
        String tunnelId = resp.getTunnelId();
        System.out.println("Create Tunnel, Id: " + tunnelId);
        //3. Define a custom callback to start automatic data consumption. 
        // TunnelWorkerConfig provides more advanced parameters. 
        TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
        TunnelWorker worker = new TunnelWorker(tunnelId, tunnelClient, config);
        try {
            worker.connectAndWorking();
        } catch (Exception e) {
            e.printStackTrace();
            config.shutdown();
            worker.shutdown();
            tunnelClient.shutdown();
        }
    }
}