Zestawienie zmian w usłudze Azure Cosmos DB dla bazy danych Apache Cassandra

DOTYCZY: Kasandra

Obsługa zestawienia zmian w usłudze Azure Cosmos DB dla systemu Apache Cassandra jest dostępna za pośrednictwem predykatów zapytań w języku Cassandra Query Language (CQL). Korzystając z tych warunków predykatu, możesz wykonać zapytanie dotyczące interfejsu API zestawienia zmian. Aplikacje mogą pobierać zmiany wprowadzone w tabeli przy użyciu klucza podstawowego (znanego również jako klucz partycji), co jest wymagane w języku CQL. Następnie możesz podjąć dalsze działania na podstawie wyników. Zmiany wierszy w tabeli są przechwytywane w kolejności ich czasu modyfikacji i kolejności sortowania na klucz partycji.

W poniższym przykładzie pokazano, jak uzyskać zestawienie zmian dla wszystkich wierszy w tabeli Cassandra Keyspace dla interfejsu API przy użyciu platformy .NET. Predykat COSMOS_CHANGEFEED_START_TIME() jest używany bezpośrednio w języku CQL do wykonywania zapytań o elementy w kanale zmian z określonej godziny rozpoczęcia (w tym przypadku bieżąca data/godzina). Pełny przykład dla języka C# można pobrać tutaj i dla języka Java tutaj.

W każdej iteracji zapytanie jest wznawiane w ostatnim punkcie, które zostały odczytane, przy użyciu stanu stronicowania. W przestrzeni kluczy można zobaczyć ciągły strumień nowych zmian w tabeli. Zobaczymy zmiany w wstawionych lub zaktualizowanych wierszach. Obserwowanie operacji usuwania przy użyciu zestawienia zmian w interfejsie API dla rozwiązania Cassandra nie jest obecnie obsługiwane.

Uwaga

Ponowne użycie tokenu po usunięciu kolekcji, a następnie ponowne utworzenie jej o tej samej nazwie powoduje wystąpienie błędu. Zalecamy ustawienie parametru pageState na wartość null podczas tworzenia nowej kolekcji i ponownego korzystania z nazwy kolekcji.

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

Aby uzyskać zmiany w jednym wierszu według klucza podstawowego, możesz dodać klucz podstawowy w zapytaniu. W poniższym przykładzie pokazano, jak śledzić zmiany dla wiersza, w którym "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);

Bieżące ograniczenia

Następujące ograniczenia mają zastosowanie w przypadku używania zestawienia zmian z interfejsem API dla rozwiązania Cassandra:

• Wstawianie i aktualizacje są obecnie obsługiwane. Operacja usuwania nie jest jeszcze obsługiwana. Aby obejść ten problem, można dodać znacznik miękki do wierszy, które są usuwane. Na przykład dodaj pole w wierszu o nazwie "deleted" i ustaw je na wartość "true".
• Ostatnia aktualizacja jest utrwalana, ponieważ w podstawowym interfejsie API dla bazy danych NoSQL i aktualizacji pośrednich dla jednostki nie są dostępne.

Obsługa błędów

Następujące kody błędów i komunikaty są obsługiwane w przypadku używania zestawienia zmian w interfejsie API dla rozwiązania Cassandra:

• Kod błędu HTTP 429 — gdy zestawienie zmian jest ograniczone, zwraca pustą stronę.

Następne kroki

• Zarządzanie zasobami usługi Azure Cosmos DB dla bazy danych Apache Cassandra przy użyciu szablonów usługi Azure Resource Manager