Apache Cassandra için Azure Cosmos DB'de değişiklik akışı

ŞUNLAR IÇIN GEÇERLIDIR: Cassandra

Apache Cassandra için Azure Cosmos DB'de değişiklik akışı desteği, Cassandra Sorgu Dili'ndeki (CQL) sorgu koşullarından kullanılabilir. Bu koşul koşullarını kullanarak değişiklik akışı API'sini sorgulayabilirsiniz. Uygulamalar CQL'de gerekli olduğu gibi birincil anahtarı (bölüm anahtarı olarak da bilinir) kullanarak tabloda yapılan değişiklikleri alabilir. Ardından sonuçlara göre daha fazla eylem gerçekleştirebilirsiniz. Tablodaki satırlarda yapılan değişiklikler, değişiklik süreleri ve bölüm anahtarı başına sıralama düzeni sırasıyla yakalanır.

Aşağıdaki örnek, .NET kullanarak Cassandra Keyspace tablosu için API'deki tüm satırlarda değişiklik akışı alma işlemini gösterir. COSMOS_CHANGEFEED_START_TIME() koşulu, belirtilen başlangıç saatinden (bu örnekte geçerli tarih saat) değişiklik akışındaki öğeleri sorgulamak için doğrudan CQL içinde kullanılır. C# için ve Java için tam örneği buradan indirebilirsiniz.

Her yinelemede, disk belleği durumu kullanılarak sorgu son noktadaki değişiklikler okundu. Keyspace'te tabloda yapılan yeni değişikliklerin sürekli akışını görebiliriz. Eklenen veya güncelleştirilen satırlarda değişiklikler göreceğiz. Cassandra için API'de değişiklik akışı kullanılarak silme işlemlerinin izlenmesi şu anda desteklenmiyor.

Not

Bir koleksiyonu bıraktıktan sonra bir belirteci yeniden kullanarak aynı adla yeniden oluşturmak hataya neden olur. Yeni bir koleksiyon oluştururken ve koleksiyon adını yeniden kullanırken pageState değerini null olarak ayarlamanızı öneririz.

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);
        }
    }

Değişiklikleri birincil anahtara göre tek bir satıra almak için, birincil anahtarı sorguya ekleyebilirsiniz. Aşağıdaki örnekte, "user_id = 1" olan satırdaki değişikliklerin nasıl izlendiği gösterilmektedir

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);

Geçerli sınırlamalar

Cassandra için API ile değişiklik akışı kullanılırken aşağıdaki sınırlamalar geçerlidir:

• Eklemeler ve güncelleştirmeler şu anda desteklenmektedir. Silme işlemi henüz desteklenmiyor. Geçici bir çözüm olarak, silinen satırlara geçici bir işaretçi ekleyebilirsiniz. Örneğin, satıra "deleted" adlı bir alan ekleyin ve bunu "true" olarak ayarlayın.
• NoSQL için çekirdek API'de olduğu gibi son güncelleştirme kalıcı hale getirilir ve varlığa yönelik ara güncelleştirmeler kullanılamaz.

Hata işleme

Cassandra için API'de değişiklik akışı kullanılırken aşağıdaki hata kodları ve iletiler desteklenir:

• HTTP hata kodu 429 - Değişiklik akışı hız sınırına sahip olduğunda boş bir sayfa döndürür.

Sonraki adımlar

• Azure Resource Manager şablonlarını kullanarak Apache Cassandra için Azure Cosmos DB kaynaklarını yönetme