Gegevens verplaatsen met behulp van de Azure Cosmos DB Spark-connector

Met Azure Synapse Analytics en Azure Synapse Link voor Azure Cosmos DB kunt u een cloudeigen hybride transactionele en analytische verwerking (HTAP) maken om analyses uit te voeren over uw gegevens in Azure Cosmos DB for NoSQL. Deze verbinding maakt integratie mogelijk via uw gegevenspijplijn aan beide uiteinden van uw gegevenswereld, Azure Cosmos DB en Azure Synapse Analytics.

Instellingen

Controleer eerst of Synapse Link is ingeschakeld op accountniveau. Dit kan worden bereikt met behulp van Azure Portal of met behulp van de Azure CLI:

az cosmosdb create --name <name> --resource-group <resource-group> --enable-analytical-storage true

U kunt ook Azure PowerShell gebruiken:

New-AzCosmosDBAccount -ResourceGroupName <resource-group> -Name <name>  -Location <location> -EnableAnalyticalStorage true

Wanneer u een container maakt, moet u analytische opslag op containerniveau per container inschakelen. Dit kan ook worden bereikt met de portal.

Dit kan ook worden bereikt met de CLI:

az cosmosdb sql container create --resource-group <resource-group> --account <account> --database <database> --name <name> --partition-key-path <partition-key-path> --throughput <throughput> --analytical-storage-ttl -1

Of met Azure PowerShell:

New-AzCosmosDBSqlContainer -ResourceGroupName <resource-group> -AccountName <account> -DatabaseName <database> -Name <name> -PartitionKeyPath <partition-key-path> -Throughput <throughput> -AnalyticalStorageTtl -1

Lezen uit Azure Cosmos DB

Er zijn twee opties om query's uit te voeren op gegevens uit Azure Cosmos DB for NoSQL. Eerst kunt u ervoor kiezen om te laden in een Spark DataFrame waarin de metagegevens in de cache worden opgeslagen. In dit voorbeeld wordt Python gebruikt om een Spark DataFrame te laden dat verwijst naar een Azure Cosmos DB for NoSQL-account.

productsDataFrame = spark.read.format("cosmos.olap")\
    .option("spark.synapse.linkedService", "cosmicworks_serv")\
    .option("spark.cosmos.container", "products")\
    .load()

U kunt ook een Spark-tabel maken die rechtstreeks naar De Azure Cosmos DB voor NoSQL verwijst. Vervolgens kunt u SparkSQL-query's uitvoeren op de Spark-tabel zonder dat dit van invloed is op het onderliggende archief. In dit voorbeeld wordt Python gebruikt om een Spark-tabel te maken.

create table products_qry using cosmos.olap options (
    spark.synapse.linkedService 'cosmicworks_serv',
    spark.cosmos.container 'products'
)

Schrijven naar Azure Cosmos DB

Als we nieuwe gegevens willen schrijven naar Azure Cosmos DB vanuit ons Spark DataFrame, kunnen we het volgende Python-script gebruiken om de gegevens in een DataFrame toe te voegen aan een bestaande container.

productsDataFrame.write.format("cosmos.oltp")\
    .option("spark.synapse.linkedService", "cosmicworks_serv")\
    .option("spark.cosmos.container", "products")\
    .mode('append')\
    .save()

We kunnen het zelfs verder doen en gegevens streamen vanuit een DataFrame, beginnend vanaf een controlepunt. We kunnen deze streaminggegevens ook toevoegen aan een bestaande container met behulp van dit Python-voorbeeldscript.

query = productsDataFrame\
    .writeStream\
    .format("cosmos.oltp")\
    .option("spark.synapse.linkedService", "cosmicworks_serv")\
    .option("spark.cosmos.container", "products")\
    .option("checkpointLocation", "/tmp/runIdentifier/")\
    .outputMode("append")\
    .start()

query.awaitTermination()