Zelfstudie: Azure Data Lake Storage Gen2, Azure Databricks en Spark

Artikel
09/28/2021
7 minuten om te lezen

In deze zelfstudie ziet u hoe u een Azure Databricks-cluster kunt verbinden met gegevens die zijn opgeslagen in een Azure-opslagaccount waarvoor Azure Data Lake Storage Gen2 is ingeschakeld. Deze verbinding stelt u in staat om systeemeigen query’s en analyses van uw cluster uit te voeren op uw gegevens.

In deze zelfstudie leert u het volgende:

Een Databricks-cluster maken
Niet-gestructureerde gegevens opnemen in een opslagaccount
Analyse uitvoeren op gegevens in Blob-opslag

Als u nog geen abonnement op Azure hebt, maak dan een gratis account aan voordat u begint.

Vereisten

Een Azure Data Lake Storage Gen2-account maken.

Zie Een opslagaccount maken dat met Azure Data Lake Storage Gen2 wordt gebruikt.
Zorg ervoor dat aan uw gebruikersaccount de rol van Gegevensbijdrager voor opslagblob is toegewezen.
Installeer AzCopy v10. Zie Gegevens overdragen met AzCopy v10
Een service-principal maken. Raadpleeg Uitleg: Gebruik de portal voor het maken van een Azure AD-toepassing en service-principal die toegang hebben tot resources.

Er zijn een paar specifieke zaken die u moet doen terwijl u de stappen in het artikel uitvoert.

✔️ Wanneer u de stappen uitvoert in de sectie De toepassing toewijzen aan een rol van het artikel, moet u ervoor zorgen dat de rol Gegevensbijdrager voor opslagblob is toegewezen aan de service-principal.

Belangrijk

Zorg ervoor dat u de rol toewijst in het bereik van het Data Lake Storage Gen2-opslagaccount. U kunt een rol toewijzen aan de bovenliggende resourcegroep of het bovenliggende abonnement, maar u ontvangt machtigingsgerelateerde fouten tot die roltoewijzingen zijn doorgegeven aan het opslagaccount.

✔️ Als u de stappen gaat uitvoeren in de sectie Waarden ophalen voor het aanmelden van het artikel, plakt u de waarden van de tenant-id, de app-id en het clientgeheim in een tekstbestand. U hebt deze binnenkort nodig.

De vluchtgegevens downloaden

In deze zelfstudie worden vluchtgegevens van het Bureau of Transport Statistics gebruikt om te laten zien hoe u een ETL-bewerking kunt uitvoeren. U moet deze gegevens downloaden om de zelfstudie te kunnen voltooien.

Ga naar Research and Innovative Technology Administration, Bureau of Transportation Statistics.
Schakel het selectievakje Prezipped file in om alle gegevensvelden te selecteren.
Selecteer de knop Download en sla de resultaten op uw computer op.
Pak de inhoud van het zipbestand uit en noteer de bestandsnaam en het pad van het bestand. U hebt deze informatie in een latere stap nodig.

Een Azure Databricks-service maken

In dit gedeelte gaat u een Azure Databricks-service maken met behulp van de Azure-portal.

Selecteer in Azure Portal Een resource maken > Analyse > Azure Databricks.

Geef bij Azure Databricks Service de volgende waarden op voor het maken van een Databricks-service:

Eigenschap	Beschrijving
Werkruimtenaam	Geef een naam op voor de Databricks-werkruimte.
Abonnement	Selecteer uw Azure-abonnement in de vervolgkeuzelijst.
Resourcegroep	Geef aan of u een nieuwe resourcegroep wilt maken of een bestaande groep wilt gebruiken. Een resourcegroep is een container met gerelateerde resources voor een Azure-oplossing. Zie Overzicht van Azure Resource Manager voor meer informatie.
Locatie	Selecteer VS - west 2. Zie Producten beschikbaar per regio voor andere beschikbare regio's.
Prijscategorie	selecteer Standaard.

Een Azure Databricks-werkruimte maken

Het duurt enkele minuten om het account te maken. Bekijk de voortgangsbalk bovenaan om de bewerkingsstatus te volgen.
Selecteer Vastmaken aan dashboard en selecteer Maken.

Een Apache Spark-cluster in Azure Databricks maken

Ga in de Azure-portal naar de Databricks-service die u hebt gemaakt en selecteer Werkruimte starten.
U wordt omgeleid naar de Azure Databricks-portal. Klik in de portal op Cluster.
Op de pagina Nieuw cluster geeft u de waarden op waarmee een nieuw cluster wordt gemaakt.

Vul de waarden voor de volgende velden in (en laat bij de overige velden de standaardwaarden staan):
- Voer een naam in voor het cluster.
- Zorg ervoor dat u het selectievakje Beëindigen na 120 minuten van inactiviteit inschakelt. Geef een duur (in minuten) op waarna het cluster moet worden beëindigd als het niet wordt gebruikt.
Selecteer Cluster maken. Als het cluster wordt uitgevoerd, kunt u notitieblokken koppelen aan het cluster en Apache Spark-taken uitvoeren.

Gegevens opnemen

Brongegevens kopiëren naar het opslagaccount

Gebruik AzCopy om gegevens uit uw csv-bestand te kopiëren naar uw Data Lake Storage Gen2-account.

Open een opdrachtpromptvenster en voer de volgende opdracht in om u aan te melden bij uw opslagaccount.
```
azcopy login
```
Volg de instructies die worden weergegeven in het opdrachtpromptvenster om uw gebruikersaccount te verifiëren.
Voer de volgende opdracht in om gegevens kopiëren vanuit het csv-account.
```
azcopy cp "<csv-folder-path>" https://<storage-account-name>.dfs.core.windows.net/<container-name>/folder1/On_Time.csv
```
- Vervang de waarde van de tijdelijke plaatsaanduiding <csv-folder-path> door het pad naar het csv-bestand.
- Vervang de waarde van de tijdelijke plaatsaanduiding <storage-account-name> door de naam van uw opslagaccount.
- Vervang de tijdelijke plaatsaanduiding <container-name> door de naam van een container in uw opslagaccount.

Een container maken en koppelen

In deze sectie maakt u een container en een map in uw opslagaccount.

Ga in de Azure-portal naar de Azure Databricks-service die u hebt gemaakt en selecteer Werkruimte starten.
Selecteer aan de linkerkant Werkruimte. Selecteer in de Werkruimte-vervolgkeuzelijst, Notitieblok > maken.
Voer in het dialoogvenster Notitieblok maken een naam voor het notitieblok in. Selecteer Python als taal en selecteer vervolgens het Apache Spark-cluster dat u eerder hebt gemaakt.
Selecteer Maken.

Kopieer en plak het volgende codeblok in de eerste cel, maar voer deze code nog niet uit.

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/<tenant>/oauth2/token",
       "fs.azure.createRemoteFileSystemDuringInitialization": "true"}

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

In dit codeblok vervangt u de tijdelijke aanduidingen appId, clientSecret, tenant en storage-account-name door de waarden die u hebt verzameld bij het uitvoeren van de vereiste stappen voor deze zelfstudie. Vervang de waarde van de tijdelijke plaatsaanduiding container-name door de naam van de container.
Druk op de toetsen Shift + Enter om de code in dit blok uit te voeren.

Houd dit notitieblok open, want u gaat er later opdrachten aan toevoegen.

Databricks Notebook gebruiken om CSV te converteren naar Parquet

Voeg aan het notitieblok dat u eerder hebt gemaakt een nieuwe cel toe en plak de volgende code in die cel.

# Use the previously established DBFS mount point to read the data.
# create a data frame to read data.

flightDF = 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.
flightDF.write.mode("append").parquet("/mnt/flightdata/parquet/flights")
print("Done")

Gegevens verkennen

Plak de volgende code in een nieuwe cel om een lijst op te halen met CSV-bestanden die zijn geüpload via AzCopy.

import os.path
import IPython
from pyspark.sql import SQLContext
display(dbutils.fs.ls("/mnt/flightdata"))

Als u een nieuw bestand en een nieuwe lijst met bestanden wilt maken in de map parquet/flights, voert u dit script uit:

dbutils.fs.put("/mnt/flightdata/1.txt", "Hello, World!", True)
dbutils.fs.ls("/mnt/flightdata/parquet/flights")

Met deze codevoorbeelden hebt u de hiërarchische aard van HDFS verkend met behulp van gegevens die zijn opgeslagen in een opslagaccount met Azure Data Lake Storage Gen2.

Query’s uitvoeren voor de gegevens

Hierna kunt u beginnen met het doorzoeken van de gegevens die u hebt geüpload in het opslagaccount. Voer elk van de volgende codeblokken in Cmd 1 in en druk op Cmd + Enter om het Python-script uit te voeren.

Als u dataframes wilt maken voor uw gegevensbronnen, voert u het volgende script uit:

Vervang de waarde van de tijdelijke plaatsaanduiding <csv-folder-path> door het pad naar het csv-bestand.

# Copy this into a Cmd cell in your notebook.
acDF = spark.read.format('csv').options(
    header='true', inferschema='true').load("/mnt/flightdata/On_Time.csv")
acDF.write.parquet('/mnt/flightdata/parquet/airlinecodes')

# read the existing parquet file for the flights database that was created earlier
flightDF = spark.read.format('parquet').options(
    header='true', inferschema='true').load("/mnt/flightdata/parquet/flights")

# print the schema of the dataframes
acDF.printSchema()
flightDF.printSchema()

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

# show the first 20 rows (20 is the default)
# to show the first n rows, run: df.show(n)
acDF.show(100, False)
flightDF.show(20, False)

# Display to run visualizations
# preferably run this in a separate cmd cell
display(flightDF)

Voer dit script uit om enkele basisquery's op de gegevens uit te voeren.

# Run each of these queries, preferably in a separate cmd cell for separate analysis
# create a temporary sql view for querying flight information
FlightTable = spark.read.parquet('/mnt/flightdata/parquet/flights')
FlightTable.createOrReplaceTempView('FlightTable')

# create a temporary sql view for querying airline code information
AirlineCodes = spark.read.parquet('/mnt/flightdata/parquet/airlinecodes')
AirlineCodes.createOrReplaceTempView('AirlineCodes')

# using spark sql, query the parquet file to return total flights in January and February 2016
out1 = spark.sql("SELECT * FROM FlightTable WHERE Month=1 and Year= 2016")
NumJan2016Flights = out1.count()
out2 = spark.sql("SELECT * FROM FlightTable WHERE Month=2 and Year= 2016")
NumFeb2016Flights = out2.count()
print("Jan 2016: ", NumJan2016Flights, " Feb 2016: ", NumFeb2016Flights)
Total = NumJan2016Flights+NumFeb2016Flights
print("Total flights combined: ", Total)

# List out all the airports in Texas
out = spark.sql(
    "SELECT distinct(OriginCityName) FROM FlightTable where OriginStateName = 'Texas'")
print('Airports in Texas: ', out.show(100))

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

Resources opschonen

Verwijder de resourcegroep en alle gerelateerde resources, wanneer u deze niet meer nodig hebt. Hiervoor selecteert u de resourcegroep voor het opslagaccount en selecteert u Verwijderen.

Volgende stappen

Gegevens extraheren, transformeren en laden met Apache Hive in Azure HDInsight