موجز التغيير في Azure Cosmos DB ل Apache Cassandra

ينطبق على: كاساندرا

يتوفر دعم موجز التغيير في Azure Cosmos DB ل Apache Cassandra من خلال دالات تقييم الاستعلام في لغة استعلام Cassandra (CQL). باستخدام هذه الشروط الأصلية، يمكنك الاستعلام عن تغيير موجز API. يمكن للتطبيقات الحصول على التغييرات التي تم إجراؤها على الجدول باستخدام المفتاح الأساسي (المعروف أيضًا باسم مفتاح القسم) كما هو مطلوب في CQL. يمكنك بعد ذلك اتخاذ مزيد من الإجراءات بناءً على النتائج. يتم تسجيل التغييرات التي تم إجراؤها على الصفوف في الجدول بترتيب وقت التعديل، وترتيب الفرز لكل مفتاح قسم.

يوضح المثال التالي كيفية الحصول على موجز تغيير على جميع الصفوف في واجهة برمجة التطبيقات لجدول Cassandra Keyspace باستخدام .NET. يتم استخدام المسند COSMOS_CHANGEFEED_START_TIME () مباشرة داخل CQL؛ للاستعلام عن العناصر في موجز التغيير من وقت بدء محدد (في هذه الحالة التاريخ والوقت الحالي). يمكنك تنزيل النموذج الكامل لـ #C ⁧هنا⁧⁩ ولجافا Java ⁧⁩هنا⁧⁩.

في كل تكرار، يُستأنف الاستعلام في آخر نقطة تمت قراءتها باستخدام حالة الترحيل. يمكننا أن نرى دفقًا مستمرًا من التغييرات الجديدة على الجدول في Keyspace. سنرى تغييرات في الصفوف التي تم إدراجها أو تحديثها. مراقبة عمليات الحذف باستخدام موجز التغيير في واجهة برمجة التطبيقات ل Cassandra غير مدعوم حاليا.

ملاحظة

إعادة استخدام رمز مميز بعد إسقاط مجموعة ثم إعادة إنشائها بنفس الاسم يؤدي إلى حدوث خطأ. ننصحك بتعيين pageState إلى 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);
        }
    }

من أجل الحصول على التغييرات على صف واحد بواسطة المفتاح الأساسي، يمكنك إضافة المفتاح الأساسي في الاستعلام. يوضح المثال التالي كيفية تتبع التغييرات للصف حيث "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);

القيود الحالية

تنطبق القيود التالية عند استخدام موجز التغيير مع واجهة برمجة التطبيقات ل Cassandra:

• الإدخالات والتحديثات مدعومة حاليًا. عملية الحذف غير معتمدة حتى الآن. كحل بديل، يمكنك إضافة علامة ناعمة على الصفوف التي يتم حذفها. على سبيل المثال: أضف حقلاً في الصف يسمى "محذوف"، واضبطه على "صحيح".
• يستمر التحديث الأخير كما هو الحال في واجهة برمجة التطبيقات الأساسية ل NoSQL ولا تتوفر التحديثات الوسيطة للكيان.

معالجة الأخطاء

يتم دعم رموز الخطأ والرسائل التالية عند استخدام موجز التغيير في واجهة برمجة التطبيقات ل Cassandra:

• ⁧⁩ رمز خطأ HTTP 429 ⁧⁩ - عندما يكون موجز التغيير محدودًا، يتم إرجاع صفحة فارغة.

الخطوات التالية

• إدارة Azure Cosmos DB لموارد Apache Cassandra باستخدام قوالب Azure Resource Manager