Rychlý start: Správa dat pomocí konektoru OLTP pro Azure Cosmos DB Spark 3 pro SQL API

PLATÍ PRO: SQL API

Tento kurz je úvodní příručkou, která ukazuje, jak pomocí konektoru Spark pro Cosmos DB číst nebo zapisovat do Cosmos DB. Cosmos DB Spark Connector je založený na Sparku 3.1.x.

V tomto rychlém kurzu se spoléháme na Azure Databricks Runtime 8.0 se Sparkem 3.1.1 a Jupyter Notebook a ukážeme si, jak používat konektor Spark pro Cosmos DB.

Můžete také použít jakoukoli jinou nabídku Sparku 3.1.1 Sparku. Také byste měli být schopni používat libovolný jazyk podporovaný Sparkem (PySpark, Scala, Java atd.) nebo jakékoli rozhraní Sparku, které už znáte (Jupyter Notebook, Livy atd.).

Požadavky

• Aktivní účet Azure. Pokud žádný nemáte, můžete si zaregistrovat bezplatný účet. Alternativně můžete použít Azure Cosmos DB Emulator pro vývoj a testování.

• Azure Databricks runtime 8.0 se Sparkem 3.1.1.

• (Volitelné) Vazba SLF4J slouží k přidružení konkrétního protokolovací architektury k SLF4J.

SLF4J je potřeba jenom v případě, že máte v plánu používat protokolování, stáhněte si také vazbu SLF4J, která propoji s rozhraním API SLF4J s implementací protokolování podle vašeho výběru. Další informace najdete v uživatelské příručce SLF4J.

Nainstalujte Cosmos DB Spark Connector do clusteru spark azure-cosmos-spark_3-1_2-12-4.3.1.jar.

Příručka Začínáme je založená na PySparku, ale můžete použít také ekvivalentní verzi Scala a následující fragment kódu můžete spustit v poznámkovém bloku Azure Databricks PySpark.

Vytváření databází a kontejnerů

Nejprve nastavte Cosmos účtu databáze a název a název kontejneru Cosmos DB.

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,
}

V dalším kroku můžete pomocí nového rozhraní API katalogu vytvořit databázi Cosmos DB a kontejner prostřednictvím Sparku.

# 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))

Při vytváření kontejnerů pomocí rozhraní API katalogu můžete nastavit propustnost a cestu ke klíči oddílu pro kontejner, který se má vytvořit.

Další informace najdete v úplné dokumentaci k rozhraní API katalogu.

Ingestace dat

Název zdroje dat je a následující příklad ukazuje, jak můžete napsat datový rámec paměti sestávající ze dvou položek do cosmos.oltp Cosmos DB:

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()

Všimněte id si, že je povinné pole pro Cosmos DB.

Další informace týkající se ingestování dat najdete v úplné dokumentaci ke konfiguraci zápisu.

Dotazování dat

Pomocí stejného cosmos.oltp zdroje dat se můžeme dotazovat na data a používat k filter nabízení filtrů:

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()

Další informace týkající se dotazování dat najdete v úplné dokumentaci ke konfiguraci dotazů.

Odvození schématu

Při dotazování dat může konektor Spark odvodit schéma na základě vzorkování existujících položek nastavením spark.cosmos.read.inferSchema.enabled na true .

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

Alternativně můžete předat vlastní schéma, které chcete použít ke čtení dat:

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()

Pokud není zadané žádné vlastní schéma a odvození schématu je zakázané, výsledná data budou vracet nezpracovaný obsah JSON položek:

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

Další informace týkající se odvozování schématu najdete v úplné dokumentaci ke konfiguraci odvozování schémat.

Referenční informace ke konfiguraci

Obecná konfigurace

Dodatečné ladění

Zápis konfigurace

Konfigurace dotazů

Konfigurace odvozování schémat

Při provádění operací čtení mohou uživatelé zadat vlastní schéma nebo povolit konektoru, aby ho odvodit. Odvozování schématu je ve výchozím nastavení povolené.

Konfigurace serializace

Slouží k ovlivnění chování serializace a deserializace JSON.

Změnit kanál (pouze pro Spark-Streaming používající cosmos.oltp.changeFeed zdroj dat, který je jen pro čtení)

Konfigurace převodu JSON

Konfigurace strategie dělení

Konfigurace řízení propustnosti

Další kroky

• Azure Cosmos DB Apache Spark 3 OLTP konektor pro jádro (SQL) API: poznámky k verzi a prostředky
• Přečtěte si další informace o Apache Sparku.