Feed di modifiche in Azure Cosmos DB for Apache Cassandra

SI APPLICA A: Cassandra

Il supporto del feed di modifiche in Azure Cosmos DB for Apache Cassandra è disponibile tramite i predicati di query nel linguaggio di query Cassandra (CQL). Usando queste condizioni di predicato, è possibile eseguire query sull'API del feed di modifiche. Le applicazioni possono ottenere le modifiche apportate a una tabella usando la chiave primaria (nota anche come chiave di partizione) come richiesto in CQL. È quindi possibile eseguire altre azioni in base ai risultati. Le modifiche apportate alle righe nella tabella vengono acquisite in base ai tempi di modifica e all'ordinamento per chiave di partizione.

L'esempio seguente illustra come ottenere un feed di modifiche in tutte le righe di una tabella nel keyspace dell'API per Cassandra usando .NET. Il predicato COSMOS_CHANGEFEED_START_TIME() viene usato direttamente all'interno di CQL per eseguire query sugli elementi nel feed di modifiche a partire da un'ora di inizio specificata (in questo caso, il valore di datetime corrente). È possibile scaricare l'esempio completo per C# qui e per Java qui.

In ogni iterazione, la query riprende in corrispondenza dell'ultimo punto in cui sono state lette le modifiche, usando lo stato di paging. È possibile visualizzare un flusso continuo di nuove modifiche apportate alla tabella nel keyspace. Verranno visualizzate le modifiche apportate alle righe inserite o aggiornate. La verifica delle operazioni di eliminazione tramite feed di modifiche nell'API per Cassandra non è attualmente supportata.

Nota

Se si riutilizza un token dopo l'eliminazione di una raccolta e quindi lo si crea nuovamente con lo stesso nome, viene restituito un errore. Quando si crea una nuova raccolta e si riutilizza il nome della raccolta, è consigliabile impostare pageState su 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);
        }
    }

Per ottenere le modifiche apportate a una singola riga per chiave primaria, è possibile aggiungere la chiave primaria nella query. L'esempio seguente mostra come tenere traccia delle modifiche per la riga dove "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);

Limitazioni correnti

Quando si usa il feed di modifiche con l'API per Cassandra, sono applicabili le limitazioni seguenti:

• Gli inserimenti e gli aggiornamenti sono attualmente supportati, ma l'operazione di eliminazione non è ancora supportata. Come soluzione alternativa, è possibile aggiungere un soft marker sulle righe da eliminare. Ad esempio, aggiungere un campo "eliminata" nella riga e impostarlo su "true".
• L'ultimo aggiornamento rimane persistente come nell'API principale per NoSQL e gli aggiornamenti intermedi all'entità non sono disponibili.

Gestione degli errori

Quando si usa il feed di modifiche nell'API per Cassandra, sono supportati i codici e i messaggi di errore seguenti:

• Codice errore HTTP 429 : quando la frequenza del feed di modifiche è limitata, viene restituita una pagina vuota.

Passaggi successivi

• Gestire le risorse di Azure Cosmos DB for Apache Cassandra con modelli di Azure Resource Manager