Quickstart: Gegevens beheren met Azure Cosmos DB Spark 3 OLTP-connector voor SQL API

VAN TOEPASSING OP: SQL-API

Deze zelfstudie is een snelstartgids om te laten zien hoe u met Cosmos DB Spark-connector kunt lezen van of schrijven naar Cosmos DB. Cosmos DB Spark-connector is gebaseerd op Spark 3.1.x.

In deze snelle zelfstudie vertrouwen we op Azure Databricks Runtime 8.0 met Spark 3.1.1 en een Jupyter Notebook om te laten zien hoe u de Cosmos DB Spark-connector gebruikt.

U kunt ook elke andere Spark 3.1.1 Spark-aanbieding gebruiken. U moet ook elke taal kunnen gebruiken die wordt ondersteund door Spark (PySpark, Scala, Java, enzovoort) of een Spark-interface die u kent (Jupyter Notebook, Livy, enzovoort).

Vereisten

• Een actief Azure-account. Als u nog geen account hebt, kunt u zich aanmelden voor een gratis account. U kunt ook de use-Azure Cosmos DB Emulator voor ontwikkeling en testen.

• Azure Databricks runtime 8.0 met Spark 3.1.1.

• (Optioneel) SLF4J-binding wordt gebruikt om een specifiek framework voor logboekregistratie te koppelen aan SLF4J.

SLF4J is alleen nodig als u logboekregistratie wilt gebruiken. Download ook een SLF4J-binding, die de SLF4J-API koppelt aan de implementatie van de logboekregistratie van uw keuze. Zie de SLF4J-gebruikershandleiding voor meer informatie.

Installeer Cosmos DB Spark-connector in uw Spark-cluster azure-cosmos-spark_3-1_2-12-4.3.1.jar

De aan de slag-handleiding is gebaseerd op PySpark, maar u kunt ook de equivalente scala-versie gebruiken en u kunt het volgende codefragment uitvoeren in een Azure Databricks PySpark-notebook.

Databases en containers maken

Stel eerst de Cosmos DB in en de Cosmos DB databasenaam en containernaam.

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

Vervolgens kunt u de nieuwe Catalog-API gebruiken om een Cosmos DB database en container te maken 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))

Wanneer u containers maakt met de Catalog-API, kunt u het doorvoer- en partitiesleutelpad instellen voor de container die moet worden gemaakt.

Zie de volledige catalogi-API-documentatie voor meer informatie.

Gegevens opnemen

De naam van de gegevensbron is en in het volgende voorbeeld ziet u hoe u een geheugengegevensframe kunt schrijven dat bestaat uit twee items die cosmos.oltp u 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()

Houd er id rekening mee dat een verplicht veld voor Cosmos DB.

Zie de volledige documentatie voor schrijfconfiguratie voor meer informatie over het opnemen van gegevens.

Querygegevens

Met dezelfde gegevensbron kunnen we gegevens cosmos.oltp opvragen en gebruiken om filter filters omlaag te pushen:

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

Zie de volledige documentatie over queryconfiguratie voor meer informatie over het uitvoeren van query's op gegevens.

Schema-deductie

Bij het uitvoeren van query's op gegevens kan de Spark-connector het schema afleiden op basis van het nemen van steekproeven van bestaande items door in te spark.cosmos.read.inferSchema.enabled stellen op true .

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

U kunt ook het aangepaste schema doorgeven dat u wilt gebruiken om de gegevens te lezen:

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

Als er geen aangepast schema is opgegeven en schemadeferentie is uitgeschakeld, retourneren de resulterende gegevens de onbewerkte JSON-inhoud van de items:

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

Zie de volledige documentatie voor de configuratie van schemadeferentie voor meer informatie met betrekking tot schemadeferentie.

Configuratieverwijzing

Algemene configuratie

Extra afstemming

Configuratie schrijven

Query-configuratie

Configuratie van schemadeferentie

Bij het uitvoeren van leesbewerkingen kunnen gebruikers een aangepast schema opgeven of de connector toestaan om dit af te lezen. Schemadeferentie is standaard ingeschakeld.

Serialisatie-configuratie

Wordt gebruikt om het json-serialisatie-/deserialisatiegedrag te beïnvloeden

Feed wijzigen (alleen voor Spark-Streaming van cosmos.oltp.changeFeed een gegevensbron, die alleen-lezen is)

Configuratie van JSON-conversie

Configuratie van partitioneringsstrategie

Configuratie van doorvoerbeheer

Volgende stappen

• Azure Cosmos DB Apache Spark 3 OLTP Connector for Core (SQL) API: Opmerkingen bij de release en resources
• Meer informatie over Apache Spark.