O feed de alterações no Azure Cosmos DB for Apache Cassandra

APLICA-SE AO: Cassandra

O suporte a Feed de alterações no Azure Cosmos DB for Apache Cassandra está disponível por meio dos predicados de consulta no CQL (Cassandra Query Language). Usando essas condições de predicado, você pode consultar a API do feed de alterações. Os aplicativos podem obter as alterações feitas em uma tabela usando a chave primária (também conhecida como chave de partição), conforme necessário no CQL. Você pode executar ações adicionais com base nos resultados. As alterações nas linhas na tabela são capturadas na ordem de seu tempo de modificação e a ordem de classificação por chave de partição.

O exemplo a seguir mostra como obter um feed de alterações em todas as linhas em uma tabela do Keyspace da API para Cassandra usando o .NET. O predicado COSMOS_CHANGEFEED_START_TIME() é usado diretamente no CQL para consultar itens no feed de alterações a partir de uma hora de início especificada (neste caso, datetime atual). Você pode baixar o exemplo completo para C# aqui e para Java aqui.

Em cada iteração, a consulta é retomada no último ponto em que as alterações foram lidas, usando o estado de paginação. Podemos ver um fluxo contínuo de novas alterações na tabela no Keyspace. Veremos as alterações nas linhas inseridas ou atualizadas. No momento, não há suporte para a observação de operações de exclusão usando feed de alteração na API para Cassandra.

Observação

Reutilizar um token depois de descartar uma coleção e recriá-lo com o mesmo nome resulta em um erro. Recomendamos que você defina o pageState como nulo ao criar uma nova coleção e reutilizar o nome da coleção.

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

Para obter as alterações em uma única linha por chave primária, você pode adicionar a chave primária na consulta. O exemplo a seguir mostra como controlar as alterações para a linha em que "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);

Limitações atuais

As seguintes limitações são aplicáveis ao usar feed de alterações com a API para Cassandra:

• Atualmente, há suporte para inserções e atualizações. As operação excluídas ainda não tem suporte. Como alternativa, você pode adicionar um marcador flexível nas linhas que estão sendo excluídas. Por exemplo, adicione um campo na linha chamada "deleted" e defina-o como "true".
• A última atualização é mantida como na API para NoSQL principal e as atualizações intermediárias para a entidade não estão disponíveis.

Tratamento de erros

Os seguintes códigos de erro e mensagens são suportados ao usar o feed de alterações na API para Cassandra:

• Código de erro HTTP 429 – quando o feed de alterações é limitado por taxa, ele retorna uma página vazia.

Próximas etapas

• Gerenciar recursos do Azure Cosmos DB for Apache Cassandra usando modelos do Azure Resource Manager