DynamoDB - Hoạt động trên bảng

Các luồng DynamoDB cho phép bạn theo dõi và phản hồi các thay đổi của mục trong bảng. Sử dụng chức năng này để tạo một ứng dụng đáp ứng các thay đổi bằng cách cập nhật thông tin giữa các nguồn. Đồng bộ dữ liệu cho hàng nghìn người dùng của một hệ thống lớn, nhiều người dùng. Sử dụng nó để gửi thông báo cho người dùng về các bản cập nhật. Các ứng dụng của nó tỏ ra đa dạng và đáng kể. Các luồng DynamoDB đóng vai trò là công cụ chính được sử dụng để đạt được chức năng này.

Các luồng nắm bắt các trình tự được sắp xếp theo thời gian chứa các sửa đổi mục trong bảng. Họ giữ dữ liệu này trong tối đa 24 giờ. Các ứng dụng sử dụng chúng để xem các mục gốc và các mục đã sửa đổi, gần như trong thời gian thực.

Các luồng được bật trên bảng ghi lại tất cả các sửa đổi. Trên bất kỳ hoạt động CRUD nào, DynamoDB tạo một bản ghi luồng với các thuộc tính khóa chính của các mục được sửa đổi. Bạn có thể định cấu hình luồng để biết thêm thông tin như hình ảnh trước và sau.

Luồng mang hai bảo đảm -

Mỗi bản ghi xuất hiện một lần trong luồng và
Mỗi sửa đổi mục dẫn đến các bản ghi luồng có cùng thứ tự với các sửa đổi.

Tất cả các luồng xử lý theo thời gian thực để cho phép bạn sử dụng chúng cho các chức năng liên quan trong các ứng dụng.

Quản lý luồng

Khi tạo bảng, bạn có thể bật luồng. Các bảng hiện có cho phép tắt luồng hoặc thay đổi cài đặt. Luồng cung cấp tính năng hoạt động không đồng bộ, có nghĩa là không ảnh hưởng đến hiệu suất bảng.

Sử dụng bảng điều khiển Quản lý AWS để quản lý luồng đơn giản. Đầu tiên, điều hướng đến bảng điều khiển và chọnTables. Trong tab Tổng quan, hãy chọnManage Stream. Bên trong cửa sổ, chọn thông tin được thêm vào luồng sửa đổi dữ liệu bảng. Sau khi nhập tất cả các cài đặt, hãy chọnEnable.

Nếu bạn muốn tắt mọi luồng hiện có, hãy chọn Manage Stream, và sau đó Disable.

Bạn cũng có thể sử dụng APIs CreateTable và UpdateTable để bật hoặc thay đổi luồng. Sử dụng tham số StreamSpecification để định cấu hình luồng. StreamEnabled chỉ định trạng thái, nghĩa là đúng đối với đã bật và sai đối với đã tắt.

StreamViewType chỉ định thông tin được thêm vào luồng: KEYS_ONLY, NEW_IMAGE, OLD_IMAGE và NEW_AND_OLD_IMAGES.

Đọc trực tuyến

Đọc và xử lý các luồng bằng cách kết nối với một điểm cuối và thực hiện các yêu cầu API. Mỗi luồng bao gồm các bản ghi luồng và mọi bản ghi tồn tại dưới dạng một sửa đổi duy nhất sở hữu luồng. Bản ghi luồng bao gồm một số thứ tự tiết lộ thứ tự xuất bản. Bản ghi thuộc về nhóm còn được gọi là mảnh. Các mảnh có chức năng như các thùng chứa cho một số bản ghi và cũng chứa thông tin cần thiết để truy cập và duyệt qua các bản ghi. Sau 24 giờ, các bản ghi sẽ tự động xóa.

Các Shard này được tạo và xóa khi cần thiết và không tồn tại lâu. Chúng cũng tự động phân chia thành nhiều phân đoạn mới, thường là để phản ứng với việc ghi tăng đột biến hoạt động. Khi tắt luồng, đóng các phân đoạn đang mở. Mối quan hệ phân cấp giữa các phân đoạn có nghĩa là các ứng dụng phải ưu tiên các phân đoạn mẹ để có thứ tự xử lý chính xác. Bạn có thể sử dụng Kinesis Adapter để tự động thực hiện việc này.

Note - Các hoạt động dẫn đến không thay đổi không ghi các bản ghi luồng.

Việc truy cập và xử lý hồ sơ yêu cầu thực hiện các tác vụ sau:

Xác định ARN của luồng mục tiêu.
Xác định (các) phân đoạn của luồng chứa các bản ghi đích.
Truy cập (các) phân đoạn để truy xuất các bản ghi mong muốn.

Note- Nên có tối đa 2 quy trình đọc một phân đoạn cùng một lúc. Nếu nó vượt quá 2 quá trình, thì nó có thể tiết lưu nguồn.

Các hành động API luồng có sẵn bao gồm

ListStreams
DescribeStream
GetShardIterator
GetRecords

Bạn có thể xem lại ví dụ sau về việc đọc luồng -

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreamsClient;

import com.amazonaws.services.dynamodbv2.model.AttributeAction;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.AttributeValueUpdate;

import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.DescribeStreamRequest;
import com.amazonaws.services.dynamodbv2.model.DescribeStreamResult;
import com.amazonaws.services.dynamodbv2.model.DescribeTableResult;

import com.amazonaws.services.dynamodbv2.model.GetRecordsRequest;
import com.amazonaws.services.dynamodbv2.model.GetRecordsResult;
import com.amazonaws.services.dynamodbv2.model.GetShardIteratorRequest;
import com.amazonaws.services.dynamodbv2.model.GetShardIteratorResult;

import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.Record;

import com.amazonaws.services.dynamodbv2.model.Shard;
import com.amazonaws.services.dynamodbv2.model.ShardIteratorType;
import com.amazonaws.services.dynamodbv2.model.StreamSpecification;
import com.amazonaws.services.dynamodbv2.model.StreamViewType;
import com.amazonaws.services.dynamodbv2.util.Tables;

public class StreamsExample {
   private static AmazonDynamoDBClient dynamoDBClient =  
      new AmazonDynamoDBClient(new ProfileCredentialsProvider());  
   private static AmazonDynamoDBStreamsClient streamsClient =  
      new AmazonDynamoDBStreamsClient(new ProfileCredentialsProvider());  

   public static void main(String args[]) {  
      dynamoDBClient.setEndpoint("InsertDbEndpointHere");   
      streamsClient.setEndpoint("InsertStreamEndpointHere");    
      
      // table creation 
      String tableName = "MyTestingTable";  
      ArrayList<AttributeDefinition> attributeDefinitions =  
         new ArrayList<AttributeDefinition>();  
      
      attributeDefinitions.add(new AttributeDefinition()
         .withAttributeName("ID") 
         .withAttributeType("N"));
         
      ArrayList<KeySchemaElement> keySchema = new 
         ArrayList<KeySchemaElement>(); 
      
      keySchema.add(new KeySchemaElement() 
         .withAttributeName("ID") 
         .withKeyType(KeyType.HASH));                       //Partition key

      StreamSpecification streamSpecification = new StreamSpecification(); 
      streamSpecification.setStreamEnabled(true); 
      streamSpecification.setStreamViewType(StreamViewType.NEW_AND_OLD_IMAGES);  
      CreateTableRequest createTableRequest = new CreateTableRequest() 
         .withTableName(tableName) 
         .withKeySchema(keySchema) 
         .withAttributeDefinitions(attributeDefinitions) 
         .withProvisionedThroughput(new ProvisionedThroughput() 
         .withReadCapacityUnits(1L) 
         .withWriteCapacityUnits(1L))
         .withStreamSpecification(streamSpecification);  
      
      System.out.println("Executing CreateTable for " + tableName); 
      dynamoDBClient.createTable(createTableRequest);  
      System.out.println("Creating " + tableName); 
      
      try { 
         Tables.awaitTableToBecomeActive(dynamoDBClient, tableName); 
      } catch (InterruptedException e) { 
         e.printStackTrace(); 
      } 
         
      // Get the table's stream settings 
      DescribeTableResult describeTableResult =
         dynamoDBClient.describeTable(tableName);  
      
      String myStreamArn = describeTableResult.getTable().getLatestStreamArn(); 
      StreamSpecification myStreamSpec =  
         describeTableResult.getTable().getStreamSpecification();  
      
      System.out.println("Current stream ARN for " + tableName + ": "+ myStreamArn);
      System.out.println("Stream enabled: "+ myStreamSpec.getStreamEnabled()); 
      System.out.println("Update view type: "+ myStreamSpec.getStreamViewType());  
      
      // Add an item 
      int numChanges = 0; 
      System.out.println("Making some changes to table data"); 
      Map<String, AttributeValue> item = new HashMap<String, AttributeValue>(); 
      item.put("ID", new AttributeValue().withN("222")); 
      item.put("Alert", new AttributeValue().withS("item!")); 
      dynamoDBClient.putItem(tableName, item); 
      numChanges++;  
      
      // Update the item         
      Map<String, AttributeValue> key = new HashMap<String, AttributeValue>(); 
      key.put("ID", new AttributeValue().withN("222")); 
      Map<String, AttributeValueUpdate> attributeUpdates =  
      new HashMap<String, AttributeValueUpdate>(); 
      
      attributeUpdates.put("Alert", new AttributeValueUpdate() 
         .withAction(AttributeAction.PUT) 
         .withValue(new AttributeValue().withS("modified item"))); 
      
      dynamoDBClient.updateItem(tableName, key, attributeUpdates); 
      numChanges++;   
      
      // Delete the item         
      dynamoDBClient.deleteItem(tableName, key);  
      numChanges++;
      
      // Get stream shards         
      DescribeStreamResult describeStreamResult =  
      streamsClient.describeStream(new DescribeStreamRequest() 
         .withStreamArn(myStreamArn)); 
      String streamArn =  
         describeStreamResult.getStreamDescription().getStreamArn(); 
      List<Shard> shards =  
         describeStreamResult.getStreamDescription().getShards();  
      
      // Process shards 
      for (Shard shard : shards) { 
         String shardId = shard.getShardId(); 
         System.out.println("Processing " + shardId + " in "+ streamArn);  
         
         // Get shard iterator 
         GetShardIteratorRequest getShardIteratorRequest = new 
            GetShardIteratorRequest() 
            .withStreamArn(myStreamArn) 
            .withShardId(shardId) 
            .withShardIteratorType(ShardIteratorType.TRIM_HORIZON); 
         
         GetShardIteratorResult getShardIteratorResult =  
            streamsClient.getShardIterator(getShardIteratorRequest); 
         String nextItr = getShardIteratorResult.getShardIterator();  
         
         while (nextItr != null && numChanges > 0) { 
            // Read data records with iterator                 
            GetRecordsResult getRecordsResult =  
               streamsClient.getRecords(new GetRecordsRequest(). 
               withShardIterator(nextItr));
               
            List<Record> records = getRecordsResult.getRecords(); 
            System.out.println("Pulling records...");  
               
            for (Record record : records) { 
               System.out.println(record); 
               numChanges--;
            } 
            nextItr = getRecordsResult.getNextShardIterator(); 
         } 
      } 
   } 
}