Kanál změn ve službě Azure Cosmos DB for Apache Cassandra

PLATÍ PRO: Cassandra

Podpora kanálu změn ve službě Azure Cosmos DB for Apache Cassandra je dostupná prostřednictvím predikátů dotazů v jazyce CQL (Cassandra Query Language). Pomocí těchto predikátových podmínek můžete dotazovat rozhraní API kanálu změn. Aplikace můžou získat změny provedené v tabulce pomocí primárního klíče (označovaného také jako klíč oddílu), jak je vyžadováno v CQL. Na základě výsledků pak můžete provést další akce. Změny řádků v tabulce se zaznamenávají v pořadí podle času jejich úpravy a pořadí řazení podle klíče oddílu.

Následující příklad ukazuje, jak pomocí .NET získat kanál změn na všech řádcích v tabulce rozhraní API pro Cassandra Keyspace. Predikát COSMOS_CHANGEFEED_START_TIME() se používá přímo v CQL k dotazování položek v kanálu změn ze zadaného počátečního času (v tomto případě aktuálního data a času). Úplnou ukázku pro C# si můžete stáhnout tady a pro Javu tady.

V každé iteraci se dotaz obnoví v posledním bodě, kdy byly přečteny změny pomocí stavu stránkování. V prostoru klíčů vidíme nepřetržitý proud nových změn v tabulce. Uvidíme změny vložených nebo aktualizovaných řádků. Sledování operací odstranění pomocí kanálu změn v rozhraní API pro Cassandra se v současné době nepodporuje.

Poznámka

Opětovném použitím tokenu po vyřazení kolekce a jeho opětovném vytvoření se stejným názvem dojde k chybě. Při vytváření nové kolekce a opakovaném použádání názvu kolekce doporučujeme nastavit pageState na hodnotu 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);
        }
    }

Pokud chcete získat změny v jednom řádku podle primárního klíče, můžete do dotazu přidat primární klíč. Následující příklad ukazuje, jak sledovat změny pro řádek, kde "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);

Aktuální omezení

Při použití kanálu změn s rozhraním API pro Cassandra platí následující omezení:

• Vložení a aktualizace jsou v současné době podporovány. Operace odstranění se zatím nepodporuje. Jako alternativní řešení můžete na řádky, které se odstraňují, přidat měkkou značku. Do řádku například přidejte pole s názvem "odstraněno" a nastavte ho na true.
• Poslední aktualizace se zachovají jako v základním rozhraní API pro NoSQL a zprostředkující aktualizace entity nejsou k dispozici.

Zpracování chyb

Při použití kanálu změn v rozhraní API pro Cassandra se podporují následující kódy chyb a zprávy:

• Kód chyby HTTP 429 – Pokud je kanál změn omezený, vrátí prázdnou stránku.

Další kroky

• Správa prostředků Azure Cosmos DB for Apache Cassandra pomocí šablon Azure Resource Manager