Spark ile Cosmos DB verilerini sorgulama

  • 5 dakika

Analiz deponuzun etkinleştirildiği Azure Cosmos DB veritabanınız için bağlı bir hizmet ekledikten sonra, Azure Synapse Analytics çalışma alanınızda bir Spark havuzu kullanarak verileri sorgulamak için bu hizmeti kullanabilirsiniz.

Azure Cosmos DB analiz verilerini bir veri çerçevesine yükleme

Azure Cosmos DB bağlı hizmetinden verilerin ilk keşfi veya hızlı analizi için, pyspark (Spark'a özgü Python uygulaması) veya Scala (genellikle Spark'ta kullanılan Java tabanlı bir dil) gibi Spark destekli bir dil kullanarak kapsayıcıdan veri çerçevesine veri yüklemek genellikle en kolay yollardan biridir.

Örneğin, aşağıdaki PySpark kodu my_linked_service bağlı hizmetini kullanarak bağlı my-container kapsayıcısı içindeki verilerden df adlı bir veri çerçevesini yüklemek ve ilk 10 veri satırını görüntülemek için kullanılabilir:

 df = spark.read
     .format("cosmos.olap")\
     .option("spark.synapse.linkedService", "my_linked_service")\
     .option("spark.cosmos.container", "my-container")\
     .load()

display(df.limit(10))

My-container kapsayıcısının aşağıdaki örneğe benzer öğeleri depolamak için kullanıldığını varsayalım:

{
    "productID": 123,
    "productName": "Widget",
    "id": "7248f072-11c3-42b1-a368-...",
    "_rid": "mjMaAL...==",
    "_self": "dbs/mjM...==/colls/mjMaAL...=/docs/mjMaAL...==/",
    "_etag": "\"54004b09-0000-2300-...\"",
    "_attachments": "attachments/",
    "_ts": 1655414791
}

PySpark kodundan elde edilen çıkış aşağıdaki tabloya benzer olacaktır:

_rid _ts productID Productname kimlik _etag
mjMaAL...== 1655414791 123 Arabirim öğesi 7248f072-11c3-42b1-a368-... 54004b09-0000-2300-...
mjMaAL...== 1655414829 124 Wotsit dc33131c-65c7-421a-a0f7-... 5400ca09-0000-2300-...
mjMaAL...== 1655414835 125 Thingumy ce22351d-78c7-428a-a1h5-... 5400ca09-0000-2300-...
... ... ... ... ... ...

Veriler işlem deposundan değil kapsayıcıdaki analiz deposundan yüklenir; operasyonel depoda sorgu ek yükü olmadığından emin olun. Analitik veri deposundaki alanlar, uygulama tanımlı alanları (bu örnekte productID ve productName) ve otomatik olarak oluşturulan meta veri alanlarını içerir.

Veri çerçevesini yükledikten sonra, verileri keşfetmek için yerel yöntemlerini kullanabilirsiniz. Örneğin aşağıdaki kod, productName tarafından sıralanmış yalnızca productID ve productName sütunlarını içeren yeni bir veri çerçevesi oluşturur:

products_df = df.select("productID", "productName").orderBy("productName")

display(products_df.limit(10))

Bu kodun çıktısı bu tabloya benzer olacaktır:

productID Productname
125 Thingumy
123 Arabirim öğesi
124 Wotsit
... ...

Cosmos DB kapsayıcısına veri çerçevesi yazma

Çoğu HTAP senaryosunda analiz deposundan Spark'a veri okumak için bağlı hizmeti kullanmanız gerekir. Ancak, aşağıdaki örnekte gösterildiği gibi bir veri çerçevesinin içeriğini kapsayıcıya yazabilirsiniz:

mydf.write.format("cosmos.oltp")\
    .option("spark.synapse.linkedService", "my_linked_service")\
    .option("spark.cosmos.container", "my-container")\
    .mode('append')\
    .save()

Dekont

Kapsayıcıya veri çerçevesi yazmak işletimsel depoyu güncelleştirir ve performansı üzerinde bir etkisi olabilir. Değişiklikler daha sonra analiz deposuna eşitlenir.

Azure Cosmos DB analiz verilerini sorgulamak için Spark SQL kullanma

Spark SQL, bir Spark havuzunda SQL dil söz dizimi ve ilişkisel veritabanı semantiği sağlayan bir Spark API'dir. Sql kullanılarak sorgulanabilen tablolara yönelik meta verileri tanımlamak için Spark SQL'i kullanabilirsiniz.

Örneğin, aşağıdaki kod önceki örneklerde kullanılan varsayımsal kapsayıcıyı temel alarak Products adlı bir tablo oluşturur:

%%sql

-- Create a logical database in the Spark metastore
CREATE DATABASE mydb;

USE mydb;

-- Create a table from the Cosmos DB container
CREATE TABLE products using cosmos.olap options (
    spark.synapse.linkedService 'my_linked_service',
    spark.cosmos.container 'my-container'
);

-- Query the table
SELECT productID, productName
FROM products;

Bahşiş

Kodun %%sql başındaki anahtar sözcük, Spark havuzuna kodu varsayılan dil (genellikle PySpark olarak ayarlanmıştır) yerine SQL olarak çalıştırmasını belirten bir büyüdür .

Bu yaklaşımı kullanarak Spark havuzunuzda, Azure Cosmos DB hesabınızdaki işletimsel depoyu etkilemeden veri analizini ve raporlama iş yüklerini desteklemek üzere Azure Cosmos DB'deki analitik verileri sorgulamak için kullanabileceğiniz bir mantıksal veritabanı oluşturabilirsiniz.