Änderungsfeed in Azure Cosmos DB for Apache Cassandra

GILT FÜR: Cassandra

Die Unterstützung für den Änderungsfeed in Azure Cosmos DB for Apache Cassandra steht über die Abfrageprädikate in der Cassandra Query Language (CQL) zur Verfügung. Mit diesen Prädikatsbedingungen können Sie die Änderungsfeed-API abfragen. Anwendungen können die an einer Tabelle vorgenommenen Änderungen mithilfe des Primärschlüssels (auch als Partitionsschlüssel bezeichnet) abrufen, der in CQL erforderlich ist. Sie können dann anhand der Ergebnisse weitere Maßnahmen ergreifen. Änderungen an den Zeilen in der Tabelle werden in der Reihenfolge ihres Änderungszeitpunkts und die Sortierreihenfolge pro Partitionsschlüssel erfasst.

Im folgenden Beispiel wird gezeigt, wie Sie einen Änderungsfeed für alle Zeilen in einer API für Cassandra-Keyspace-Tabelle mithilfe von .NET abrufen. Das Prädikat COSMOS_CHANGEFEED_START_TIME() wird direkt in CQL verwendet, um Elemente im Änderungsfeed ab einer angegebenen Startzeit (in diesem Fall das aktuelle Datum und die Uhrzeit) abzufragen. Sie können das ganze Beispiel für C# hier und für Java hier herunterladen.

In jeder Iterationen wird die Abfrage mithilfe des Pagingzustands am letzten Punkt fortgesetzt, an dem Änderungen gelesen wurden. Sie können einen kontinuierlichen Stream neuer Änderungen an der Tabelle im Keyspace erkennen. Es werden Änderungen an Zeilen angezeigt, die eingefügt oder aktualisiert wurden. Die Überwachung auf Löschvorgänge mithilfe des Änderungsfeeds wird in der API für Cassandra derzeit nicht unterstützt.

Hinweis

Wenn Sie ein Token nach dem Löschen einer Sammlung wiederverwenden und dann mit demselben Namen neu erstellen, tritt ein Fehler auf. Es empfiehlt sich, den pageState-Wert beim Erstellen einer neuen Sammlung auf NULL festzulegen und den Namen der Sammlung erneut zu verwenden.

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

Um die Änderungen an einer einzelnen Zeile nach Primärschlüssel abzurufen, können Sie der Abfrage den Primärschlüssel hinzufügen. Im folgenden Beispiel wird gezeigt, wie Sie Änderungen für die Zeile mit „user_id = 1“ nachverfolgen.

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

Aktuelle Einschränkungen

Es gelten die folgenden Einschränkungen, wenn der Änderungsfeed mit der API für Cassandra verwendet wird:

• Einfügungen und Aktualisierungen werden zurzeit unterstützt. Löschvorgänge werden noch nicht unterstützt. Als Problemumgehung können Sie den gelöschten Zeilen eine schwache Markierung hinzufügen. Fügen Sie z. B. in der Zeile ein Feld mit dem Namen „deleted“ hinzu, und legen Sie es auf „true“ fest.
• Das letzte Update wird wie in der API für NoSQl beibehalten, und zwischengeschaltete Aktualisierungen der Entität sind nicht verfügbar.

Fehlerbehandlung

Die folgenden Fehlercodes und -meldungen werden bei der Verwendung des Änderungsfeeds in der API für Cassandra unterstützt:

• HTTP-Fehlercode 429: Wenn der Änderungsfeed gedrosselt ist, wird eine leere Seite zurückgegeben.

Nächste Schritte

• Verwalten von Azure Cosmos DB for Apache Cassandra-Ressourcen mithilfe von Azure Resource Manager-Vorlagen