Azure Cosmos DB for Apache Cassandra 變更摘要

適用於： Cassandra

Azure Cosmos DB for Apache Cassandra 中的變更摘要支援可透過 Cassandra 查詢語言 (CQL) 中的查詢述詞來取得。 您可以使用這些述詞條件來查詢變更摘要 API。 應用程式可以使用主索引鍵 (也稱為分割區索引鍵) 來取得對資料表所做的變更，因為它在 CQL 中是必要項目。 您接著可以根據結果來採取進一步動作。 資料表中資料列的變更會依其修改時間的順序以及每個分割區索引鍵的排序順序來進行擷取。

下列範例顯示如何使用 .NET 來取得 API for Cassandra Keyspace 資料表中所有資料列的變更摘要。 COSMOS_CHANGEFEED_START_TIME() 述詞直接用於 CQL 內，以從指定的開始時間 (在此案例中為目前日期時間) 來查詢變更摘要中的項目。 您可以在這裡 (針對 C#) 和 這裡 (針對 Java) 下載完整範例。

在每個反覆項目中，會使用分頁狀態，在上次讀取變更的最後一個位置繼續查詢。 我們可以在 Keyspace 中看到新資料表變更的連續資料流。 我們將會看到已插入或更新的資料列變更。 目前不支援在 API for Cassandra 中使用變更摘要來監看刪除作業。

注意

在卸除集合之後重複使用權杖，然後使用相同名稱重新建立權杖會導致錯誤。 建議您在建立新的集合並重複使用集合名稱時，將 pageState 設定為 Null。

Java

    Session cassandraSession = utils.getSession();

    try {
        DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");  
        LocalDateTime now = LocalDateTime.now().minusHours(6).minusMinutes(30);  
        String query="SELECT * FROM uprofile.user where COSMOS_CHANGEFEED_START_TIME()='" 
            + dtf.format(now)+ "'";
        
        byte[] token=null; 
        System.out.println(query); 
        while(true)
        {
            SimpleStatement st=new  SimpleStatement(query);
            st.setFetchSize(100);
            if(token!=null)
                st.setPagingStateUnsafe(token);
            
            ResultSet result=cassandraSession.execute(st) ;
            token=result.getExecutionInfo().getPagingState().toBytes();
            
            for(Row row:result)
            {
                System.out.println(row.getString("user_name"));
            }
        }
    } finally {
        utils.close();
        LOGGER.info("Please delete your table after verifying the presence of the data in portal or from CQL");
    }

C#

    //set initial start time for pulling the change feed
     DateTime timeBegin = DateTime.UtcNow;

    //initialise variable to store the continuation token
    byte[] pageState = null;
    while (true)
    {
        try
        {

            //Return the latest change for all rows in 'user' table    
            IStatement changeFeedQueryStatement = new SimpleStatement(
            $"SELECT * FROM uprofile.user where COSMOS_CHANGEFEED_START_TIME() = '{timeBegin.ToString("yyyy-MM-ddTHH:mm:ss.fffZ", CultureInfo.InvariantCulture)}'");
            if (pageState != null)
            {
                changeFeedQueryStatement = changeFeedQueryStatement.SetPagingState(pageState);
            }
            Console.WriteLine("getting records from change feed at last page state....");
            RowSet rowSet = session.Execute(changeFeedQueryStatement);

            //store the continuation token here
            pageState = rowSet.PagingState;

            List<Row> rowList = rowSet.ToList();
            if (rowList.Count != 0)
            {
                for (int i = 0; i < rowList.Count; i++)
                {
                    string value = rowList[i].GetValue<string>("user_name");
                    int key = rowList[i].GetValue<int>("user_id");
                    // do something with the data - e.g. compute, forward to another event, function, etc.
                    // here, we just print the user name field
                    Console.WriteLine("user_name: " + value);
                }
            }
            else
            {
                Console.WriteLine("zero documents read");
            }
        }
        catch (Exception e)
        {
            Console.WriteLine("Exception " + e);
        }
    }

若要依主索引鍵來取得單一資料列的變更，您可以在查詢中新增主索引鍵。 下列範例顯示如何追蹤資料列的變更，其中 "user_id = 1"

C#

    //Return the latest change for all row in 'user' table where user_id = 1
    IStatement changeFeedQueryStatement = new SimpleStatement(
    $"SELECT * FROM uprofile.user where user_id = 1 AND COSMOS_CHANGEFEED_START_TIME() = '{timeBegin.ToString("yyyy-MM-ddTHH:mm:ss.fffZ", CultureInfo.InvariantCulture)}'");

Java

    String query="SELECT * FROM uprofile.user where user_id=1 and COSMOS_CHANGEFEED_START_TIME()='" 
                       + dtf.format(now)+ "'";
    SimpleStatement st=new  SimpleStatement(query);

目前的限制

搭配使用變更摘要與 API for Cassandra 時，會有下列限制︰

• 目前支援插入和更新。 尚未支援刪除作業。 因應措施是您可以在要刪除的資料列上新增軟標記。 例如，在稱為 "deleted" 的資料列中新增欄位，並將它設定為 "true"。
• 最後一個更新會持續保存在核心 API for NoSQL 中，但無法使用實體的中繼更新。

錯誤處理

在 API for Cassandra 中使用變更摘要時，支援下列錯誤碼和訊息：

• HTTP 錯誤碼 429 - 對變更摘要進行速度限制時，會傳回空白頁面。

下一步

• 使用 Azure Resource Manager 範本管理 Azure Cosmos DB for Apache Cassandra 資源