Kanál změn ve službě Azure Cosmos DB pro 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 jazyce CQL. Pak můžete na základě výsledků provést další akce. Změny řádků v tabulce se zaznamenávají v pořadí jejich času úprav a pořadí řazení podle klíče oddílu.

Následující příklad ukazuje, jak získat informační kanál změn na všech řádcích v rozhraní API pro Cassandra Keyspace tabulky pomocí .NET. Predikát COSMOS_CHANGEFEED_START_TIME() se používá přímo v jazyce 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 zde a pro Javu tady.

V každé iteraci se změny dotazu v posledním okamžiku četly pomocí stavu stránkování. V keyspace vidíme nepřetržitý proud nových změn tabulky. Uvidíme změny vložených nebo aktualizovaných řádcích. 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é spuštění tokenu po vyřazení kolekce a jeho opětovné vytvoření se stejným názvem způsobí chybu. Při vytváření nové kolekce a opakovaném spuště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 jednoho řá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 řádku, 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í:

• Vkládání a aktualizace se v současné době podporují. Operace odstranění se zatím nepodporuje. Jako alternativní řešení můžete přidat obnovitelné značky na řádky, které se odstraňují. Přidejte například do řádku pole s názvem "odstraněno" a nastavte ho na true.
• Poslední aktualizace je trvalá jako v základním rozhraní API pro NoSQL a přechodné 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 pro Apache Cassandra pomocí šablon Azure Resource Manageru