Mengubah umpan di Azure Cosmos DB untuk Apache Cassandra

BERLAKU UNTUK: Cassandra

Dukungan umpan perubahan di Azure Cosmos DB for Apache Cassandra tersedia melalui predikat kueri dalam Cassandra Query Language (CQL). Dengan menggunakan kondisi predikat ini, Anda dapat melakukan permintaan API perubahan umpan. Aplikasi bisa mendapatkan perubahan terjadi pada tabel menggunakan kunci primer (juga dikenal sebagai kunci partisi) seperti yang diperlukan dalam CQL. Anda kemudian dapat mengambil tindakan lebih lanjut berdasarkan tataan hasil. Perubahan pada baris dalam tabel dianbil dalam urutan waktu modifikasinya dan susunan urutan sortir per kunci partisi.

Contoh berikut menunjukkan cara mendapatkan umpan perubahan pada semua baris dalam API untuk tabel Cassandra Keyspace menggunakan .NET. Predikat COSMOS_CHANGEFEED_START_TIME() digunakan langsung dalam CQL untuk meminta item dalam ubah umpan dari waktu mulai yang ditentukan (dalam hal ini tanggalwaktu saat ini). Anda dapat melakukan pengunduhan sampel lengkap, untuk C # di sini dan untuk Java di sini.

Dalam setiap perulangan, permintaan dilanjutkan pada titik terakhir perubahan dibaca menggunakan kondisi halaman. Kita dapat melihat stream terus menerus perubahan baru ke tabel di Keyspace. Kita akan melihat perubahan pada rekaman yang disisipkan atau diperbarui. Menonton operasi penghapusan menggunakan umpan perubahan di API untuk Cassandra saat ini tidak didukung.

Catatan

Menggunakan kembali token setelah menghapus koleksi dan kemudian membuatnya kembali dengan nama yang sama akan menghasilkan kesalahan. Kami menyarankan Anda untuk mengatur pageState ke null saat membuat koleksi baru dan menggunakan kembali nama koleksi.

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

Untuk mendapatkan perubahan pada satu rekaman menurut kunci primer, Anda bisa menambahkan kunci primer dalam permintaan. Contoh yang mengikuti menunjukkan cara melacak perubahan untuk rekaman di mana "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);

Batasan saat ini

Batasan berikut berlaku saat menggunakan umpan perubahan dengan API untuk Cassandra:

• Sisipkan dan pembarui saat ini didukung. Operasi menghapus belum didukung. Sebagai solusi sementara, Anda bisa menambahkan penanda lembut pada item yang sedang dihapus. Misalnya, tambahkan bidang di rekaman yang disebut "dihapus" dan set ke "true".
• Pembaruan terakhir dipertahankan seperti dalam API inti untuk NoSQL dan pembaruan perantara untuk entitas tidak tersedia.

Penanganan kesalahan

Kode kesalahan dan pesan berikut didukung saat menggunakan umpan perubahan di API untuk Cassandra:

• Kode galat HTTP 429 - Saat perubahan umpan dibatasi, maka akan ditampilkan halaman kosong.

Langkah berikutnya

• Mengelola sumber daya Azure Cosmos DB untuk Apache Cassandra menggunakan templat Azure Resource Manager