Snabbstart: Hantera data med Azure Cosmos DB Spark 3 OLTP-anslutningsapp för SQL API

GÄLLER FÖR: SQL API

Den här självstudien är en snabbstartsguide som visar hur du använder Cosmos DB Spark-anslutningsapp för att läsa från eller skriva till Cosmos DB. Cosmos DB Spark-anslutningsappen baseras på Spark 3.1.x.

I den här snabbskursen förlitar vi oss på Azure Databricks Runtime 8.0 med Spark 3.1.1 och en Jupyter Notebook som visar hur du använder Cosmos DB Spark Connector.

Du kan även använda andra Spark 3.1.1 Spark-erbjudanden. Du bör också kunna använda alla språk som stöds av Spark (PySpark, Scala, Java osv.) eller alla Spark-gränssnitt som du är bekant med (Jupyter Notebook, Livy osv.).

Förutsättningar

• Ett aktivt Azure-konto. Om du inte har ett kan du registrera dig för en kostnadsfri utvärderingsversion. Du kan också använda -Azure Cosmos DB Emulator för utveckling och testning.

• Azure Databricks runtime 8.0 med Spark 3.1.1.

• (Valfritt) SLF4J-bindning används för att associera ett specifikt loggningsramverk med SLF4J.

SLF4J behövs bara om du planerar att använda loggning. Du kan även ladda ned en SLF4J-bindning som länkar SLF4J-API:et med valfri loggningsimplementering. Mer information finns i användarhandboken för SLF4J.

Installera Cosmos DB Spark Connector i spark-klustret azure-cosmos-spark_3-1_2-12-4.3.1.jar

Kom igång-guiden baseras på PySpark, men du kan även använda motsvarande scala-version och du kan köra följande kodfragment i en Azure Databricks PySpark-anteckningsbok.

Skapa databaser och containrar

Ange först Cosmos DB-autentiseringsuppgifter och Cosmos DB databasnamn och containernamn.

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

Sedan kan du använda det nya katalog-API:et för att skapa Cosmos DB databas och container via Spark.

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

När du skapar containrar med katalog-API:et kan du ange dataflödes- och partitionsnyckelsökvägen för containern som ska skapas.

Mer information finns i den fullständiga katalogens API-dokumentation.

Mata in data

Namnet på datakällan är , och i följande exempel visas hur du kan skriva en minnesdataram som består av två cosmos.oltp objekt som ska 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()

Observera att id är ett obligatoriskt fält för Cosmos DB.

Mer information om inmatning av data finns i den fullständiga dokumentationen för skrivkonfiguration.

Söka i data

Med hjälp av cosmos.oltp samma datakälla kan vi köra frågor mot data och använda för filter att skicka ned 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()

Mer information om hur du frågar efter data finns i den fullständiga dokumentationen för frågekonfiguration.

Schemaferens

När du frågar efter data kan Spark-anslutningsappen dra slutsatser om schemat baserat på sampling av befintliga objekt genom att ange spark.cosmos.read.inferSchema.enabled till true .

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

Du kan också skicka det anpassade schema som du vill ska användas för att läsa data:

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

Om inget anpassat schema har angetts och schemaferens har inaktiverats returnerar resulterande data Json-rådata för objekten:

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

Mer information om schemaferens finns i den fullständiga dokumentationen för schemareferenskonfiguration.

Konfigurationsreferens

Allmän konfiguration

Extra justering

Skrivkonfiguration

Frågekonfiguration

Konfiguration av schemareferens

När du utför läsåtgärder kan användarna ange ett anpassat schema eller tillåta att anslutningsappen härar det. Schemainferens är aktiverat som standard.

Serialiseringskonfiguration

Används för att påverka json-serialisering/deserialiseringsbeteende

Ändringsflöde (endast för Spark-Streaming med cosmos.oltp.changeFeed hjälp av datakälla, som är skrivskyddade) konfiguration

Konfiguration av Json-konvertering

Konfiguration av partitioneringsstrategi

Konfiguration av dataflödeskontroll

Nästa steg

• Azure Cosmos DB Apache Spark 3 OLTP Connector for Core (SQL) API: Viktig information och resurser
• Läs mer om Apache Spark.