Verschieben von Daten mit dem Azure Cosmos DB-Spark-Connector

Mit Azure Synapse Analytics und Azure Synapse Link für Azure Cosmos DB können Sie eine cloudeigene hybride Transaktions- und Analyseverarbeitung (HTAP) erstellen, um Analysen über Ihre Daten in Azure Cosmos DB für NoSQL auszuführen. Diese Verbindung ermöglicht die Integration über Ihre Datenpipeline an beiden Enden Ihrer Datenwelt, Azure Cosmos DB und Azure Synapse Analytics.

Einrichten

Stellen Sie zunächst sicher, dass der Synapse-Link auf Kontoebene aktiviert ist. Dazu können Sie das Azure-Portal oder die Azure CLI verwenden:

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

Sie können auch Azure PowerShell verwenden:

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

Beim Erstellen eines Containers sollten Sie den analytischen Speicher auf Containerebene für jeden einzelnen Container aktivieren. Auch dafür kann das Portal verwendet werden.

Dies ist auch mit der CLI möglich:

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

Auch Azure PowerShell kann verwendet werden:

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

Lesen aus Azure Cosmos DB

Es gibt zwei Optionen zum Abfragen von Daten aus Azure Cosmos DB for NoSQL. Sie können sich zum einen für das Laden in einen Spark-Dataframe entscheiden, in dem die Metadaten zwischengespeichert werden. In diesem Beispiel wird Python verwendet, um einen Spark-Dataframe zu laden, der auf ein Azure Cosmos DB for NoSQL-Konto verweist.

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

Alternativ können Sie eine Spark-Tabelle erstellen, die direkt auf die Azure Cosmos DB for NoSQL verweist. Anschließend können Sie SparkSQL-Abfragen für die Spark-Tabelle ausführen, ohne den zugrunde liegenden Speicher zu beeinträchtigen. In diesem Beispiel wird Python zum Erstellen einer Spark-Tabelle verwendet.

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

Schreiben in Azure Cosmos DB

Wenn Sie neue Daten aus dem Spark-Dataframe in Azure Cosmos DB schreiben möchten, können Sie das folgende Python-Skript verwenden, um die Daten in einem Dataframe an einen vorhandenen Container anzufügen.

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

Sie können sogar noch weiter gehen und Daten aus einem Dataframe ab einem Prüfpunkt streamen. Sie können diese Streamingdaten auch mithilfe des folgenden Python-Beispielskripts an einen vorhandenen Container anfügen.

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