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 önkoşulları aracılığıyla 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 bir 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 BIR API'deki tüm satırlarda değişiklik akışının nasıl alıneceğini gösterir. COSMOS_CHANGEFEED_START_TIME() koşulu, belirtilen başlangıç zamanından (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 tam örneği buradan ve Java için buradan indirebilirsiniz.

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

Not

Bir koleksiyonu bıraktıktan sonra bir belirtecin yeniden kullanılıp aynı adla yeniden oluşturma işlemi hatayla sonuçlanır. 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 sorguya birincil anahtarı 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 "silinmiş" 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ıdır ve varlıkta 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ırlı 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