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 behulp van deze predicaatvoorwaarden kunt u een query uitvoeren op de WIJZIGINGENfeed-API. Toepassingen kunnen de wijzigingen in een tabel ophalen met behulp van de primaire sleutel (ook wel de partitiesleutel genoemd), zoals vereist in CQL. U kunt vervolgens 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 items in de wijzigingenfeed op te vragen vanaf een opgegeven begintijd (in dit geval huidige datum/tijd). U kunt het volledige voorbeeld downloaden, voor C# hier en voor Java hier.

In elke iteratie wordt de query hervat op het laatste punt dat wijzigingen zijn gelezen, met behulp van de pagineringsstatus. We zien een continue stroom nieuwe wijzigingen in de tabel in de Keyspace. Er worden wijzigingen weergegeven in rijen die worden ingevoegd of bijgewerkt. Het controleren van verwijderbewerkingen met behulp van de wijzigingenfeed in DE 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. We raden u aan om pageState in te stellen op null wanneer u een nieuwe verzameling maakt en de naam van de verzameling opnieuw gebruikt.

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 bijhoudt voor de rij 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. De verwijderbewerking wordt nog niet ondersteund. Als tijdelijke oplossing kunt u een zachte markering toevoegen aan rijen die worden verwijderd. Voeg bijvoorbeeld een veld toe aan de rij met de naam 'verwijderd' en stel dit in op 'true'.
• Laatste update blijft behouden als 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 DE API voor Cassandra:

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

Volgende stappen

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