hızlı başlangıç: SQL apı için Azure Cosmos DB Spark 3 OLTP bağlayıcısı ile verileri yönetme

UYGULANDıĞı YER: SQL API

bu öğretici, Cosmos DB okuma veya yazma için Cosmos DB Spark bağlayıcısını nasıl kullanacağınızı gösteren hızlı bir başlangıç kılavuzudur. spark Connector, spark 3.1. x ' i temel alır Cosmos DB.

bu hızlı öğreticide, spark 3.1.1 ile Azure Databricks Runtime 8,0 ' i ve Cosmos DB spark bağlayıcısının nasıl kullanılacağını gösteren bir Jupyter Notebook güveniyoruz.

Başka bir Spark 3.1.1 Spark teklifini de kullanabilirsiniz. Ayrıca, Spark (PySpark, Scala, Java, vb.) tarafından desteklenen herhangi bir dili veya bildiğiniz herhangi bir Spark arabirimini (Jupyter Notebook, Livy, vb.) kullanabilirsiniz.

Önkoşullar

• Etkin bir Azure hesabı. Bir aboneliğiniz yoksa ücretsiz bir hesap için kaydolabilirsiniz. alternatif olarak, geliştirme ve test için Azure Cosmos DB kullan Emulator kullanabilirsiniz.

• Spark 3.1.1 ile çalışma zamanı 8,0 Azure Databricks .

• Seçim Dolayısıyla slf4j Binding , belirli bir günlük çerçevesini dolayısıyla slf4j ile ilişkilendirmek için kullanılır.

DOLAYıSıYLA SLF4J yalnızca, günlüğü kullanmayı planlıyorsanız gerekir, Ayrıca, DOLAYıSıYLA SLF4J API 'sini tercih ettiğiniz günlük uygulamasıyla bağlayacaktır. Daha fazla bilgi için dolayısıyla slf4j Kullanıcı kılavuzuna bakın.

spark kümenizde Cosmos DB spark bağlayıcısını ( azure-cosmos-spark_3 -1_2-12-4.3.1. jar ) yükleyip

Başlangıç Kılavuzu, PySpark 'a dayalıdır, ancak eşdeğer Scala sürümünü de kullanabilirsiniz ve aşağıdaki kod parçacığını bir Azure Databricks PySpark not defterinde çalıştırabilirsiniz.

Veritabanları ve kapsayıcılar oluşturma

ilk olarak, Cosmos DB hesabı kimlik bilgilerini ve Cosmos DB veritabanı adını ve kapsayıcı adını ayarlayın.

cosmosEndpoint = "https://REPLACEME.documents.azure.com:443/"
cosmosMasterKey = "REPLACEME"
cosmosDatabaseName = "sampleDB"
cosmosContainerName = "sampleContainer"

cfg = {
  "spark.cosmos.accountEndpoint" : cosmosEndpoint,
  "spark.cosmos.accountKey" : cosmosMasterKey,
  "spark.cosmos.database" : cosmosDatabaseName,
  "spark.cosmos.container" : cosmosContainerName,
}

daha sonra, Spark aracılığıyla bir Cosmos DB veritabanı ve kapsayıcı oluşturmak için yeni katalog apı 'sini kullanabilirsiniz.

# Configure Catalog Api to be used
spark.conf.set("spark.sql.catalog.cosmosCatalog", "com.azure.cosmos.spark.CosmosCatalog")
spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountEndpoint", cosmosEndpoint)
spark.conf.set("spark.sql.catalog.cosmosCatalog.spark.cosmos.accountKey", cosmosMasterKey)

# create a cosmos database using catalog api
spark.sql("CREATE DATABASE IF NOT EXISTS cosmosCatalog.{};".format(cosmosDatabaseName))

# create a cosmos container using catalog api
spark.sql("CREATE TABLE IF NOT EXISTS cosmosCatalog.{}.{} using cosmos.oltp TBLPROPERTIES(partitionKeyPath = '/id', manualThroughput = '1100')".format(cosmosDatabaseName, cosmosContainerName))

Katalog API 'SI ile kapsayıcılar oluştururken, oluşturulacak kapsayıcının aktarım hızını ve bölüm anahtarı yolunu ayarlayabilirsiniz.

Daha fazla bilgi için bkz. tam Katalog API 'si belgeleri.

Veriyi çekme

veri kaynağının adı cosmos.oltp ve aşağıdaki örnekte, Cosmos DB için iki öğeden oluşan bir bellek veri çerçevesini nasıl yazabileceğiniz gösterilmektedir:

spark.createDataFrame((("cat-alive", "Schrodinger cat", 2, True), ("cat-dead", "Schrodinger cat", 2, False)))\
  .toDF("id","name","age","isAlive") \
   .write\
   .format("cosmos.oltp")\
   .options(**cfg)\
   .mode("APPEND")\
   .save()

idCosmos DB için zorunlu bir alan olduğunu unutmayın.

Verilerle ilgili daha fazla bilgi almak için bkz. tam yazma yapılandırması belgeleri.

Verileri sorgulama

Aynı cosmos.oltp veri kaynağını kullanarak, filtreleri göndermek için verileri sorgulayabilir ve kullanabilirsiniz filter :

from pyspark.sql.functions import col

df = spark.read.format("cosmos.oltp").options(**cfg)\
 .option("spark.cosmos.read.inferSchema.enabled", "true")\
 .load()

df.filter(col("isAlive") == True)\
 .show()

Verileri sorgulama hakkında daha fazla bilgi için bkz. tam sorgu yapılandırması belgeleri.

Şema çıkarımı

Verileri sorgularken Spark Bağlayıcısı, üzerine ayarlayarak varolan öğeleri örneklemeye göre şemayı çıkarsçıkarabilir spark.cosmos.read.inferSchema.enabled true .

df = spark.read.format("cosmos.oltp").options(**cfg)\
 .option("spark.cosmos.read.inferSchema.enabled", "true")\
 .load()
 
df.printSchema()

Alternatif olarak, verileri okumak için kullanmak istediğiniz özel şemayı geçirebilirsiniz:

customSchema = StructType([
      StructField("id", StringType()),
      StructField("name", StringType()),
      StructField("type", StringType()),
      StructField("age", IntegerType()),
      StructField("isAlive", BooleanType())
    ])

df = spark.read.schema(schema).format("cosmos.oltp").options(**cfg)\
 .load()
 
df.printSchema()

Özel şema belirtilmemişse ve Şema çıkarımı devre dışı bırakılmışsa, sonuçta elde edilen veriler öğelerin ham JSON içeriğini döndürür:

df = spark.read.format("cosmos.oltp").options(**cfg)\
 .load()
 
df.printSchema()

Şema çıkarımı ile ilgili daha fazla bilgi için bkz. tam Şema çıkarımı yapılandırması belgeleri.

Yapılandırma başvurusu

Genel yapılandırma

Ek ayarlama

Yazma yapılandırması

Sorgu yapılandırması

Şema çıkarımı yapılandırması

Okuma işlemleri yaparken, kullanıcılar özel bir şema belirtebilir veya bağlayıcının onu çıkarması için izin verebilir. Şema çıkarımı varsayılan olarak etkindir.

Serileştirme yapılandırması

JSON serileştirme/seriden serileştirme davranışını etkilemek için kullanılır

Değişiklik akışı (yalnızca Spark-Streaming kaynağı kullanan veri cosmos.oltp.changeFeed kaynağı için) yapılandırma

Json dönüştürme yapılandırması

Bölümleme stratejisi yapılandırması

Aktarım hızı denetimi yapılandırması

Sonraki adımlar

• Azure Cosmos DB Apache Spark 3 Çekirdek için OLTP Bağlayıcısı (SQL) API'si: Sürüm notları ve kaynaklar
• Apache Spark hakkında daha fazla bilgi edinin.