Wijzigingenfeed in Azure Cosmos DB voor Apache Cassandra

VAN TOEPASSING OP: Cassandra

Ondersteuning voor wijzigingenfeeds in Azure Cosmos DB voor Apache Cassandra is beschikbaar via de querypredicaten in de Cassandra Query Language (CQL). Met deze predicaatvoorwaarden kunt u een query uitvoeren op de API voor de wijzigingenfeed. Toepassingen kunnen de wijzigingen in een tabel ophalen met behulp van de primaire sleutel (ook wel de partitiesleutel genoemd) zoals vereist is in CQL. Vervolgens kunt u verdere acties uitvoeren op basis van de resultaten. Wijzigingen in de rijen in de tabel worden vastgelegd in de volgorde van de wijzigingstijd en de sorteervolgorde per partitiesleutel.

In het volgende voorbeeld ziet u hoe u een wijzigingenfeed kunt ophalen voor alle rijen in een API voor Cassandra Keyspace-tabel met behulp van .NET. Het predicaat COSMOS_CHANGEFEED_START_TIME() wordt rechtstreeks in CQL gebruikt om vanaf een opgegeven begintijd een query uit te voeren op items in de wijzigingenfeed (in dit geval huidige datum/tijd). U kunt hier en voor Java het volledige voorbeeld downloaden, voor C# hier en voor Java.

In elke iteratie wordt de query hervat op het laatste punt dat wijzigingen zijn gelezen, met behulp van de pagingstatus. We kunnen een doorlopende stroom nieuwe wijzigingen in de tabel in de Keyspace zien. We zien wijzigingen in rijen die zijn ingevoegd of bijgewerkt. Het controleren op verwijderingsbewerkingen met behulp van wijzigingenfeed in API voor Cassandra wordt momenteel niet ondersteund.

Notitie

Het opnieuw gebruiken van een token na het verwijderen van een verzameling en het opnieuw maken met dezelfde naam resulteert in een fout. U wordt aangeraden de pageState in te stellen op null bij het maken van een nieuwe verzameling en het opnieuw gebruiken van de naam van de verzameling.

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

Als u de wijzigingen in één rij per primaire sleutel wilt ophalen, kunt u de primaire sleutel toevoegen aan de query. In het volgende voorbeeld ziet u hoe u wijzigingen voor de rij bijhoudt waarbij '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);

Huidige beperkingen

De volgende beperkingen zijn van toepassing bij het gebruik van wijzigingenfeed met API voor Cassandra:

• Invoegen en updates worden momenteel ondersteund. Verwijderbewerking wordt nog niet ondersteund. Als tijdelijke oplossing kunt u een zachte markering toevoegen aan rijen die worden verwijderd. Voeg bijvoorbeeld een veld toe in de rij met de naam 'verwijderd' en stel dit in op 'true'.
• De laatste update blijft behouden, zoals in de kern-API voor NoSQL en tussenliggende updates voor de entiteit zijn niet beschikbaar.

Foutafhandeling

De volgende foutcodes en berichten worden ondersteund bij het gebruik van wijzigingenfeed in API voor Cassandra:

• HTTP-foutcode 429 : wanneer de wijzigingenfeed beperkt is, wordt er een lege pagina geretourneerd.

Volgende stappen

• Azure Cosmos DB voor Apache Cassandra-resources beheren met behulp van Azure Resource Manager-sjablonen