Ansluta till Azure DocumentDB från Azure Databricks

Den här artikeln visar hur du ansluter till Azure DocumentDB från Azure Databricks för att utföra vanliga dataåtgärder med Hjälp av Python och Spark. Du konfigurerar nödvändiga beroenden, upprättar en anslutning och kör läs-, skriv-, filter- och aggregeringsåtgärder med MongoDB Spark-anslutningsappen.

Prerequisites

  • En prenumeration på Azure

  • Ett befintligt Azure DocumentDB-kluster

Konfigurera Azure Databricks-arbetsyta

Konfigurera din Azure Databricks-arbetsyta så att den ansluter till Azure DocumentDB. Lägg till MongoDB Connector för Spark-biblioteket i din beräkning för att aktivera anslutning till Azure DocumentDB.

  1. Gå till Azure Databricks-arbetsytan.

  2. Konfigurera standardberäkningen som är tillgänglig eller skapa en ny beräkningsresurs för att köra din notebook.

  3. Välj en Databricks-runtime som stöder minst Spark 3.0.

  4. I beräkningsresursen väljer du Bibliotek>Installera ny>Maven.

  5. Lägg till Maven-koordinaterna: org.mongodb.spark:mongo-spark-connector_2.12:3.0.1

  6. Välj Installera.

  7. Starta om beräkningen när installationen är klar.

Konfigurera anslutningsinställningar

Konfigurera Spark att använda azure DocumentDB-anslutningssträngen för alla läs- och skrivåtgärder.

  1. I Azure-portalen navigerar du till din Azure DocumentDB-resurs.

  2. Under Inställningar>Anslutningssträngar kopierar du anslutningssträngen. Den har formuläret: mongodb+srv://<user>:<password>@<database_name>.mongocluster.cosmos.azure.com

  3. I Azure Databricks går du till beräkningskonfigurationen och väljer Avancerade alternativ (längst ned på sidan).

  4. Lägg till följande Spark-konfigurationsvariabler:

    • spark.mongodb.output.uri – Klistra in anslutningssträngen
    • spark.mongodb.input.uri – Klistra in anslutningssträngen

  5. Spara konfigurationen.

Du kan också ange anslutningssträngen direkt i koden med hjälp .option() av metoden när du läser eller skriver data.

Skapa Python-notebook-fil

Kör dina dataåtgärder genom att skapa en ny Python-anteckningsbok.

  1. Skapa en ny Python-anteckningsbok på din Azure Databricks-arbetsyta.

  2. Definiera dina anslutningsvariabler i början av notebook-filen:

    connectionString = "mongodb+srv://<user>:<password>@<database_name>.mongocluster.cosmos.azure.com/?tls=true&authMechanism=SCRAM-SHA-256&retrywrites=false&maxIdleTimeMS=120000"
database = "<database_name>"
collection = "<collection_name>"

  3. Ersätt platshållarvärdena med ditt faktiska databasnamn och samlingsnamn.

Läsa data från samling

Läsa data från din Azure DocumentDB-samling till en Spark DataFrame.

  1. Använd följande kod för att läsa in data från din samling:

    df = spark.read.format("mongo").option("database", database).option("spark.mongodb.input.uri", connectionString).option("collection", collection).load()

  2. Kontrollera att data har lästs in framgångsrikt.

    df.printSchema()
display(df)

  3. Observera resultatet. Den här koden skapar en DataFrame som innehåller alla dokument från den angivna samlingen och visar schemat och data.

Filtrera data

Använd filter för att hämta specifika delmängder av data från din samling.

  1. Använd DataFrame-metoden filter() för att tillämpa villkor:

    df_filtered = df.filter(df["birth_year"] == 1970)
display(df_filtered)

  2. Använd kolumnindexnummer:

    df_filtered = df.filter(df[2] == 1970)
display(df_filtered)

  3. Observera resultatet. Den här metoden returnerar endast de dokument som matchar dina filtervillkor.

Fråga efter data med SQL

Skapa tillfälliga vyer och kör SQL-frågor mot dina data för välbekant SQL-baserad analys.

  1. Skapa en tillfällig vy från din DataFrame:

    df.createOrReplaceTempView("T")

  2. Kör SQL-frågor mot vyn:

    df_result = spark.sql("SELECT * FROM T WHERE birth_year == 1970 AND gender == 2")
display(df_result)

  3. Observera resultatet. Med den här metoden kan du använda SQL-standardsyntax för komplexa frågor och kopplingar.

Skriva data till samling

Spara nya eller ändrade data genom att skriva tillbaka DataFrames till Azure DocumentDB-samlingar.

  1. Använd följande kod för att skriva data till en samling:

    df.write.format("mongo").option("spark.mongodb.output.uri", connectionString).option("database", database).option("collection", "CitiBike2019").mode("append").save()

  2. Skrivåtgärden slutförs utan utdata. Kontrollera att skrivåtgärden har slutförts genom att läsa data från samlingen:

    df_verify = spark.read.format("mongo").option("database", database).option("spark.mongodb.input.uri", connectionString).option("collection", "CitiBike2019").load()
display(df_verify)

    Tips/Råd

    Använd olika skrivlägen, till exempel append, overwriteeller ignore beroende på dina krav.

Köra aggregeringspipelines

Kör aggregeringspipelines för att utföra databearbetning och analys på serversidan direkt i Azure DocumentDB. Aggregeringspipelines möjliggör kraftfulla datatransformeringar, gruppering och beräkningar utan att flytta data från databasen. De är idealiska för analys i realtid, instrumentpaneler och rapportgenerering.

  1. Definiera din sammansättningspipeline som en JSON-sträng:

    pipeline = "[{ $group : { _id : '$birth_year', totaldocs : { $count : 1 }, totalduration: {$sum: '$tripduration'}} }]"

  2. Kör pipelinen och läs in resultatet:

    df_aggregated = spark.read.format("mongo").option("database", database).option("spark.mongodb.input.uri", connectionString).option("collection", collection).option("pipeline", pipeline).load()
display(df_aggregated)

Relaterat innehåll

  • Uppdaterades senast