Lernprogramm: Azure Data Lake Storage, Azure Databricks und Spark

In diesem Lernprogramm erfahren Sie, wie Sie Ihr Azure Databricks Cluster mit Azure Data Lake Storage verbinden und Spark-Abfragen und Analysen für Ihre Daten ausführen. Außerdem erfahren Sie, wie Sie unstrukturierte Flugdaten aufnehmen und sie für eine effiziente Abfrage in das Apache-Parkettformat umwandeln.

In diesem Tutorial lernen Sie:

  • Unstrukturierte Daten in ein Speicherkonto aufnehmen
  • Ausführen von Analysen für Ihre Daten in Blob Storage

Wenn Sie kein Azure-Abonnement besitzen, können Sie ein kostenloses Konto erstellen, bevor Sie beginnen.

Voraussetzungen

Erstellen eines Azure Databricks-Arbeitsbereichs und eines Notebooks

  1. Erstellen eines Azure Databricks-Arbeitsbereichs Weitere Informationen finden Sie unter Erstellen eines Azure Databricks-Arbeitsbereichs.

  2. Erstellen Sie ein Notebook. Weitere Informationen finden Sie unter Erstellen eines Notebooks. Wählen Sie Python als Standardsprache für das Notebook aus.

Lassen Sie Ihr Notebook geöffnet. Es wird in den folgenden Abschnitten verwendet.

Herunterladen der Flugdaten

In diesem Tutorial werden Pünktlichkeitsdaten zu Flügen für Januar 2016 vom Bureau of Transportation Statistics verwendet, um zu zeigen, wie ein ETL-Vorgang durchgeführt wird. Sie müssen diese Daten zum Durchführen des Tutorials herunterladen.

  1. Laden Sie die Datei On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip herunter. Diese Datei enthält die Flugdaten.

  2. Entzippen Sie den Inhalt der ZIP-Datei, und notieren Sie sich den Namen und den Pfad der Datei. Diese Informationen werden in einem späteren Schritt benötigt.

Informationen über die in den Leistungsdaten zur Pünktlichkeitsberichterstattung erfassten Informationen finden Sie in den Feldbeschreibungen auf der Website des Bureau of Transportation Statistics.

Daten erfassen

In diesem Abschnitt laden Sie die CSV-Flugdaten in Ihr Azure Data Lake Storage-Konto hoch und stellen dann das Speicherkonto in Ihrem Databricks-Cluster bereit. Schließlich verwenden Sie Databricks, um die CSV-Flugdaten zu lesen und sie zurück in den Speicher im Apache-Parquet-Format zu schreiben.

Hochladen der Flugdaten in Ihr Speicherkonto

Kopieren Sie mithilfe von AzCopy Ihre CSV-Datei in Ihr Azure Data Lake Storage-Konto. Verwenden Sie den azcopy make Befehl, um einen Container in Ihrem Speicherkonto zu erstellen. Verwenden Sie dann den azcopy copy Befehl, um die csv-Daten zu kopieren, die Sie soeben in ein Verzeichnis in diesem Container heruntergeladen haben.

Geben Sie in den folgenden Schritten Namen für den Container ein, den Sie erstellen möchten, sowie für das Verzeichnis und den Blob, in die Sie die Flugdaten im Container hochladen möchten. Verwenden Sie die vorgeschlagenen Namen in jedem Schritt, oder geben Sie Ihre eigenen Namen an, während Sie den Benennungskonventionen für Container, Verzeichnisse und Blobs folgen.

  1. Öffnen Sie ein Eingabeaufforderungsfenster, und geben Sie den folgenden Befehl ein, um sich bei Microsoft Entra ID anzumelden, um auf Ihr Speicherkonto zuzugreifen.

    azcopy login

    Folgen Sie den Anweisungen im Eingabeaufforderungsfenster, um Ihr Benutzerkonto zu authentifizieren.

  2. Um einen Container in Ihrem Speicherkonto zum Speichern der Flugdaten zu erstellen, geben Sie den folgenden Befehl ein:

    azcopy make  "https://<storage-account-name>.dfs.core.windows.net/<container-name>"

    • Ersetzen Sie den Platzhalterwert <storage-account-name> durch den Namen Ihres Speicherkontos.

    • Ersetzen Sie den <container-name> Platzhalter durch einen Namen für den Container, den Sie zum Speichern der CSV-Daten erstellen möchten, z. B. flight-data-container.

  3. Um die CSV-Daten in Ihr Speicherkonto hochzuladen (kopieren), geben Sie den folgenden Befehl ein.

    azcopy copy "<csv-folder-path>" https://<storage-account-name>.dfs.core.windows.net/<container-name>/<directory-name>/On_Time.csv

    • Ersetzen Sie den Platzhalterwert <csv-folder-path> durch den Pfad zu der CSV-Datei.

    • Ersetzen Sie den Platzhalterwert <storage-account-name> durch den Namen Ihres Speicherkontos.

    • Ersetzen Sie den <container-name>-Platzhalter durch den Namen des Containers in Ihrem Speicherkonto.

    • Ersetzen Sie den <directory-name> Platzhalter durch den Namen eines Verzeichnisses, um Ihre Daten im Container zu speichern, z. B. jan2016.

Bereitstellen Ihres Speicherkontos in Ihrem Databricks-Cluster

In diesem Abschnitt stellen Sie Ihren Azure Data Lake Storage-Cloudobjektspeicher im Databricks-Dateisystem (Databricks File System, DBFS) bereit. Sie verwenden den Microsoft Entra Dienstprinzipal, den Sie zuvor für die Authentifizierung beim Speicherkonto erstellt haben. Weitere Informationen finden Sie unter Mount Cloud-Objektspeicher auf Azure Databricks.

  1. Fügen Sie Ihr Notebook an Ihren Cluster an.

    1. Wählen Sie im zuvor erstellten Notebook die Schaltfläche Verbinden in der oberen rechten Ecke der Notebook-Symbolleiste aus. Diese Schaltfläche öffnet das Compute-Auswahlmenü. (Wenn Sie Ihr Notizbuch bereits mit einem Cluster verbunden haben, wird der Name dieses Clusters im Schaltflächentext anstelle von "Verbinden") angezeigt.

    2. Wählen Sie im Dropdownmenü des Clusters alle zuvor erstellten Cluster aus.

    3. Der Text in der Cluster-Auswahl ändert sich zu Starten. Warten Sie, bis der Cluster vollständig gestartet ist und der Name des Clusters auf der Schaltfläche angezeigt wird, bevor Sie fortfahren.

  2. Kopieren Sie den folgenden Codeblock, und fügen Sie ihn in die erste Zelle ein, führen Sie den Code jedoch noch nicht aus.

    configs = {"fs.azure.account.auth.type": "OAuth",
       "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
       "fs.azure.account.oauth2.client.id": "<appId>",
       "fs.azure.account.oauth2.client.secret": "<clientSecret>",
       "fs.azure.account.oauth2.client.endpoint": "https://login.microsoftonline.com/<tenantId>/oauth2/token",
       "fs.azure.createRemoteFileSystemDuringInitialization": "true"}

dbutils.fs.mount(
source = "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<directory-name>",
mount_point = "/mnt/flightdata",
extra_configs = configs)

  3. In diesem Codeblock:

    • Ersetzen Sie in configs die Platzhalterwerte <appId>, <clientSecret> und <tenantId> durch die Anwendungs-ID, das Clientgeheimnis und die Mandanten-ID, die Sie beim Erstellen des Dienstprinzipals im Abschnitt „Voraussetzungen“ kopiert haben.

    • Ersetzen Sie im URI source die Platzhalterwerte <storage-account-name>, <container-name> und <directory-name> durch die Namen Ihres Azure Data Lake Storage-Speicherkontos, des Containers und des Verzeichnisses, die Sie beim Hochladen der Flugdaten in das Speicherkonto angegeben haben.

      Hinweis

      Der Schemabezeichner im URI, abfss, weist Databricks an, den Azure Blob File System-Treiber mit Transport Layer Security (TLS) zu verwenden. Weitere Informationen zum URI finden Sie unter Verwenden des Azure Data Lake Storage-URI.

  4. Stellen Sie sicher, dass der Cluster mit dem Starten fertig ist, bevor Sie fortfahren.

  5. Drücken Sie die Tasten SHIFT + ENTER, um den Code in diesem Block auszuführen.

Der Container und das Verzeichnis, in das Sie die Flugdaten in Ihrem Speicherkonto hochgeladen haben, ist jetzt über den Bereitstellungspunkt /mnt/flightdata in Ihrem Notebook verfügbar.

Verwenden Sie ein Databricks-Notebook, um CSV in Parquet zu konvertieren

Nachdem Sie nun über einen DBFS-Einhängepunkt auf die csv-Flugdaten zugreifen können, verwenden Sie einen Apache Spark DataFrame, um die Daten in Ihren Arbeitsbereich zu laden und sie anschließend im Apache-Parquet-Format zurück in Ihren Azure Data Lake Storage-Objektspeicher zu schreiben.

  • Ein Spark-DataFrame ist eine zweidimensionale, bezeichnete Datenstruktur mit Spalten potenziell unterschiedlicher Typen. Verwenden Sie einen DataFrame, um Daten in verschiedenen unterstützten Formaten einfach zu lesen und zu schreiben. Mit einem DataFrame können Sie Daten aus dem Cloudobjektspeicher laden und Analysen und Transformationen darin innerhalb Ihres Computeclusters durchführen, ohne dass sich dies auf die zugrunde liegenden Daten im Cloudobjektspeicher auswirkt. Weitere Informationen finden Sie unter Arbeiten mit PySpark-DataFrames in Azure Databricks.

  • Apache Parquet ist ein spaltenorientiertes Dateiformat mit Optimierungen, die Abfragen beschleunigen. Dieses Dateiformat ist effizienter als CSV oder JSON. Weitere Informationen finden Sie unter Parquet-Dateien.

Fügen Sie zunächst im Notebook eine neue Zelle hinzu und dann den folgenden Code ein.

# Use the previously established DBFS mount point to read the data.
# Create a DataFrame to read the csv data.
# The header option specifies that the first row of data should be used as the DataFrame column names
# The inferschema option specifies that the column data types should be inferred from the data in the file
flight_df = spark.read.format('csv').options(
    header='true', inferschema='true').load("/mnt/flightdata/*.csv")

# Read the airline csv file and write the output to parquet format for easy query.
flight_df.write.mode("append").parquet("/mnt/flightdata/parquet/flights")
print("Done")

Drücken Sie die Tasten UMSCHALT + EINGABETASTE, um den Code in diesem Block auszuführen.

Bevor Sie mit dem nächsten Abschnitt fortfahren, stellen Sie sicher, dass alle Parkettdaten geschrieben werden, und "Fertig" wird in der Ausgabe angezeigt.

Durchsuchen von Daten

Verwenden Sie in diesem Abschnitt das Databricks-Dateisystemprogramm, um Den Azure Data Lake Storage Objektspeicher mithilfe des DBFS-Bereitstellungspunkts zu erkunden, den Sie im vorherigen Abschnitt erstellt haben.

Fügen Sie in einer neuen Zelle den folgenden Code ein, um eine Liste der Dateien am Bereitstellungspunkt abzurufen. Der erste Befehl gibt eine Liste von Dateien und Verzeichnissen aus. Der zweite Befehl zeigt die Ausgabe im tabellarischen Format an, um das Lesen zu erleichtern.

dbutils.fs.ls("/mnt/flightdata")
display(dbutils.fs.ls("/mnt/flightdata"))

Drücken Sie die Tasten SHIFT + ENTER, um den Code in diesem Block auszuführen.

Beachten Sie, dass das Parquet-Verzeichnis in der Auflistung angezeigt wird. Sie haben die CSV-Flugdaten im Parquet-Format im Verzeichnis parquet/flights (Parquet/Flüge) im vorherigen Abschnitt gespeichert. Um Dateien im Verzeichnis parquet/flights (Parquet/Flüge) auflisten zu können, fügen Sie den folgenden Code in eine neue Zelle ein, und führen Sie ihn aus:

display(dbutils.fs.ls("/mnt/flightdata/parquet/flights"))

Um eine neue Datei zu erstellen und sie auflisten zu können, fügen Sie den folgenden Code in eine neue Zelle ein, und führen Sie ihn aus:

dbutils.fs.put("/mnt/flightdata/mydirectory/mysubdirectory/1.txt", "Hello, World!", True)
display(dbutils.fs.ls("/mnt/flightdata/mydirectory/mysubdirectory"))

Da Sie die Datei 1.txt in diesem Tutorial nicht benötigen, können Sie den folgenden Code in eine Zelle einfügen und ausführen, um mydirectory rekursiv zu löschen. Der Parameter True gibt einen rekursiven Löschvorgang an.

dbutils.fs.rm("/mnt/flightdata/mydirectory", True)

Verwenden Sie als Einfachheit den Hilfebefehl, um Details zu anderen Befehlen zu erfahren.

dbutils.fs.help("rm")

Mithilfe dieser Codebeispiele haben Sie die hierarchische Natur von HDFS mithilfe von Daten untersucht, die in einem Speicherkonto gespeichert sind, wobei Azure Data Lake Storage aktiviert sind.

Abfragen der Daten

Fragen Sie als Nächstes die Daten ab, die Sie in Ihr Speicherkonto hochgeladen haben. Geben Sie jeden der folgenden Codeblöcke in eine neue Zelle ein, und drücken Sie auf die UMSCHALTTASTE + EINGABETASTE, um das Python-Skript auszuführen.

DataFrames bieten zahlreiche Funktionen (Auswählen von Spalten, Filtern, Verknüpfen, Aggregieren), mit denen Sie häufige Probleme bei der Datenanalyse effizient beheben können.

Um ein DataFrame aus Ihren zuvor gespeicherten Parquet-Flugdaten zu laden und einige der unterstützten Funktionen zu erkunden, geben Sie dieses Skript in eine neue Zelle ein, und führen Sie es aus.

# Read the existing parquet file for the flights database that was created earlier
flight_df = spark.read.parquet("/mnt/flightdata/parquet/flights")

# Print the schema of the dataframe
flight_df.printSchema()

# Print the flight database size
print("Number of flights in the database: ", flight_df.count())

# Show the first 25 rows (20 is the default)
# To show the first n rows, run: df.show(n)
# The second parameter indicates that column lengths shouldn't be truncated (default is 20 characters)
flight_df.show(25, False)

# You can also use the DataFrame to run simple queries. Results are returned in a DataFrame.
# Show the first 25 rows of the results of a query that returns selected columns for all flights originating from airports in Texas
flight_df.select("FlightDate", "Reporting_Airline", "Flight_Number_Reporting_Airline", "OriginCityName", "DepTime", "DestCityName", "ArrTime", "ArrDelay").filter("OriginState = 'TX'").show(258, False)

# Use display to run visualizations
# Preferably run this in a separate cmd cell
display(flight_df)

Geben Sie dieses Skript in eine neue Zelle ein, um einige grundlegende Analyseabfragen für die Daten auszuführen. Sie können sich dafür entscheiden, das gesamte Skript (UMSCHALTTASTE + EINGABETASTE) auszuführen, jede Abfrage zu markieren und sie separat mit STRG + UMSCHALTTASTE + EINGABETASTE auszuführen oder jede Abfrage in eine separate Zelle einzugeben und dort auszuführen.

# create a temporary sql view for querying flight information
flight_data = spark.read.parquet('/mnt/flightdata/parquet/flights')
flight_data.createOrReplaceTempView('FlightTable')

# Print the total number of flights in Jan 2016 (the number of rows in the flight data).
print("Number of flights in Jan 2016: ", flight_data.count())

# Using spark sql, query the parquet file to return the total flights of each airline
num_flights_by_airline=spark.sql("SELECT Reporting_Airline, count(*) AS NumFlights FROM FlightTable GROUP BY Reporting_Airline ORDER BY NumFlights DESC")
num_flights_by_airline.show()

# List out all the airports in Texas
airports_in_texas = spark.sql(
    "SELECT DISTINCT(OriginCityName) FROM FlightTable WHERE OriginStateName = 'Texas'")
print('Airports in Texas: ', airports_in_texas.count())
airports_in_texas.show(100, False)

# Find all airlines that fly from Texas
airlines_flying_from_texas = spark.sql(
    "SELECT DISTINCT(Reporting_Airline) FROM FlightTable WHERE OriginStateName='Texas'")
print('Airlines that fly to/from Texas: ', airlines_flying_from_texas.count())
airlines_flying_from_texas.show(100, False)

# List airlines by average arrival delay (negative values indicate early flights)
avg_arrival_delay=spark.sql(
    "SELECT Reporting_Airline, count(*) AS NumFlights, avg(DepDelay) AS AverageDepDelay, avg(ArrDelay) AS AverageArrDelay FROM FlightTable GROUP BY Reporting_Airline ORDER BY AverageArrDelay DESC")
print("Airlines by average arrival delay")
avg_arrival_delay.show()

# List airlines by the highest percentage of delayed flights. A delayed flight is one with a  departure or arrival delay that is greater than 15 minutes
spark.sql("DROP VIEW IF EXISTS totalFlights")
spark.sql("DROP VIEW IF EXISTS delayedFlights")
spark.sql(
    "CREATE TEMPORARY VIEW totalFlights AS SELECT Reporting_Airline, count(*) AS NumFlights FROM FlightTable GROUP BY Reporting_Airline")
spark.sql(
    "CREATE TEMPORARY VIEW delayedFlights AS SELECT Reporting_Airline, count(*) AS NumDelayedFlights FROM FlightTable WHERE DepDelay>15 or ArrDelay>15 GROUP BY Reporting_Airline")
percent_delayed_flights=spark.sql(
    "SELECT totalFlights.Reporting_Airline, totalFlights.NumFlights, delayedFlights.NumDelayedFlights, delayedFlights.NumDelayedFlights/totalFlights.NumFlights*100 AS PercentFlightsDelayed FROM totalFlights INNER JOIN delayedFlights ON totalFlights.Reporting_Airline = delayedFlights.Reporting_Airline ORDER BY PercentFlightsDelayed DESC")
print("Airlines by percentage of flights delayed")
percent_delayed_flights.show()

Zusammenfassung

In diesem Tutorial erfahren Sie:

  • Erstellte Azure-Ressourcen, einschließlich eines Azure Data Lake Storage-Kontos und eines Microsoft Entra-Dienstprinzipals, und zugewiesene Berechtigungen für den Zugriff auf das Speicherkonto.

  • Sie haben einen Azure Databricks-Arbeitsbereich und ein Notebook erstellt.

  • Sie haben AzCopy verwendet, um unstrukturierte CSV-Flugdaten in das Azure Data Lake Storage-Speicherkonto hochzuladen.

  • Sie haben Funktionen des Databricks-Dateisystem-Hilfsprogramms zum Bereitstellen Ihres Azure Data Lake Storage-Speicherkontos verwendet und das hierarchische Dateisystem erkundet.

  • Sie haben Apache Spark-DataFrames verwendet, um Ihre CSV-Flugdaten in das Apache-Parquet-Format zu transformieren und sie wieder in Ihrem Azure Data Lake Storage-Speicherkonto zu speichern.

  • DataFrames verwendet, um die Flugdaten zu untersuchen und eine einfache Abfrage auszuführen.

  • Apache Spark SQL verwendet, um die Flugdaten für die Gesamtanzahl der Flüge für jede Fluggesellschaft im Januar 2016, die Flughäfen in Texas, die Fluggesellschaften, die von Texas aus fliegen, die durchschnittliche Ankunftsverzögerung in Minuten für jede Fluggesellschaft national und den Prozentsatz der Flüge jeder Fluggesellschaft, die verspätete Abflüge oder Ankünfte haben, abzufragen.

Bereinigen von Ressourcen

Um das Notebook beizubehalten und Gebühren zu vermeiden, schließen (beenden) Sie Ihren Cluster. Wählen Sie es in der Compute-Auswahl oben rechts in der Notebook-Symbolleiste aus, wählen Sie im Menü Beenden und bestätigen Sie Ihre Auswahl. (Standardmäßig wird der Cluster nach 120 Minuten Inaktivität automatisch beendet.)

Verwenden Sie die linke Randleiste des Arbeitsbereichs, um einzelne Arbeitsbereichsressourcen wie Notizbücher und Cluster zu löschen. Ausführliche Anweisungen finden Sie unter Löschen eines Clusters oder Löschen eines Notebooks.

Löschen Sie die Ressourcengruppe und alle dazugehörigen Ressourcen, wenn Sie sie nicht mehr benötigen. Wählen Sie hierzu im Azure-Portal die Ressourcengruppe für das Speicherkonto und den Arbeitsbereich und dann Löschen aus.

Nächste Schritte

Extrahieren, Transformieren und Laden von Daten mithilfe von Apache Hive in Azure HDInsight

  • Last updated on