kurz: Azure Data Lake Storage Gen2, Azure Databricks & Spark

  • Článek
  • 6 min ke čtení

v tomto kurzu se dozvíte, jak připojit cluster Azure Databricks k datům uloženým v účtu úložiště Azure, který má povolený Azure Data Lake Storage Gen2. Toto připojení umožňuje nativně spouštět dotazy a analýzy z vašeho clusteru na vašich datech.

V tomto kurzu:

  • Vytvoření clusteru Databricks
  • Ingestace nestrukturovaných dat do účtu úložiště
  • Spuštění analýzy dat v úložišti objektů BLOB

Pokud ještě nemáte předplatné Azure, vytvořte si napřed bezplatný účet.

Požadavky

Stažení letových údajů

V tomto kurzu se k předvedení operace ETL používá letová data z statistiky předsednictví. Tato data musíte stáhnout, abyste mohli kurz dokončit.

  1. Přejít na výzkum a inovativní správu technologií, statistiky dopravy.

  2. Pokud chcete vybrat všechna datová pole, zaškrtněte políčko pro soubor ve formátu ZIP .

  3. Klikněte na tlačítko Stáhnout a uložte výsledky do svého počítače.

  4. Rozbalte obsah souboru zip a poznamenejte si název souboru a cestu k souboru. Tyto informace budete potřebovat v pozdějším kroku.

Vytvoření služby Azure Databricks

V této části vytvoříte službu Azure Databricks pomocí Azure Portal.

  1. V Azure Portal vyberte vytvořit Azure Databricks prostředků > Analytics > .

    Datacihly na Azure Portal

  2. V části Azure Databricks služba zadejte následující hodnoty pro vytvoření služby datacihly:

    Vlastnost Popis
    Název pracovního prostoru Zadejte název pracovního prostoru Databricks.
    Předplatné Z rozevíracího seznamu vyberte své předplatné Azure.
    Skupina prostředků Určete, jestli chcete vytvořit novou skupinu prostředků, nebo použít existující. Skupina prostředků je kontejner, který uchovává související prostředky pro řešení Azure. Další informace naleznete v tématu Přehled skupin prostředků v Azure.
    Umístění Vyberte USA – západ 2. Další dostupné oblasti najdete v tématu Dostupné služby Azure podle oblastí.
    Cenová úroveň Vyberte Standard.

    Vytvoření pracovního prostoru Azure Databricks

  3. Vytvoření účtu trvá několik minut. Chcete-li monitorovat stav operace, zobrazte indikátor průběhu v horní části.

  4. Vyberte Připnout na řídicí panel a potom vyberte Vytvořit.

Vytvoření clusteru Spark v Azure Databricks

  1. V Azure Portal otevřete službu datacihly, kterou jste vytvořili, a vyberte Spustit pracovní prostor.

  2. Budete přesměrováni na portál Azure Databricks. Na portálu vyberte Cluster.

    Datacihly v Azure

  3. Na stránce New cluster (Nový cluster) zadejte hodnoty pro vytvoření clusteru.

    Vytvoření clusteru datacihly Spark v Azure

    Zadejte hodnoty následujících polí a potvrďte výchozí hodnoty dalších polí:

    • Zadejte název clusteru.

    • Nezapomeňte zaškrtnout políčko Terminate after 120 minutes of inactivity (Ukončit po 120 minutách nečinnosti). Zadejte dobu (v minutách), po které se má ukončit činnost clusteru, pokud se cluster nepoužívá.

  4. Vyberte vytvořit cluster. Po spuštění clusteru můžete ke clusteru připojit poznámkové bloky a spouštět úlohy Spark.

Ingestace dat

Zkopírování zdrojových dat do účtu úložiště

pomocí AzCopy zkopírujte data ze souboru .csv do svého účtu Data Lake Storage Gen2.

  1. Otevřete okno příkazového řádku a zadejte následující příkaz, který se přihlásí k účtu úložiště.

    azcopy login

    Postupujte podle pokynů, které se zobrazí v okně příkazového řádku, a ověřte svůj uživatelský účet.

  2. Chcete-li kopírovat data z .csv účtu, zadejte následující příkaz.

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

    • Nahraďte <csv-folder-path> hodnotu zástupného symbolu cestou k souboru .csv .

    • Nahraďte <storage-account-name> hodnotu zástupného symbolu názvem vašeho účtu úložiště.

    • <container-name>Zástupný symbol nahraďte názvem kontejneru v účtu úložiště.

Vytvoření kontejneru a jeho připojení

V této části vytvoříte kontejner a složku ve svém účtu úložiště.

  1. V Azure Portalotevřete službu Azure Databricks, kterou jste vytvořili, a vyberte Spustit pracovní prostor.

  2. Na levé straně vyberte pracovní prostor. V rozevíracím seznamu Pracovní prostor vyberte Vytvořit > Poznámkový blok.

    Vytvoření poznámkového bloku v datacihlech

  3. V dialogovém okně Vytvořit poznámkový blok zadejte název poznámkového bloku. Jako jazyk vyberte Python a pak vyberte cluster Spark, který jste vytvořili dříve.

  4. Vyberte Vytvořit.

  5. Zkopírujte následující blok kódu a vložte ho do první buňky, ale tento kód ještě nespouštějte.

    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)

  6. V tomto bloku kódu nahraďte appId clientSecret tenant storage-account-name zástupné hodnoty,, a v tomto bloku kódu hodnotami, které jste shromáždili při dokončování požadavků tohoto kurzu. Nahraďte container-name hodnotu zástupného symbolu názvem kontejneru.

  7. Stiskněte klávesy SHIFT + ENTER a spusťte kód v tomto bloku.

Ponechte tento poznámkový blok otevřený a později do něj přidejte příkazy.

Použití poznámkového bloku Databricks k převodu CSV na formát Parquet

V poznámkovém bloku, který jste vytvořili dříve, přidejte novou buňku a vložte do této buňky následující kód.

# 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")

Zkoumání dat

Do nové buňky vložte následující kód, který načte seznam souborů CSV odeslaných prostřednictvím AzCopy.

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

Pokud chcete vytvořit nový soubor a zobrazit seznam souborů ve složce parquet/flights, spusťte tento skript:

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

Na těchto vzorových kódech jste prozkoumali hierarchickou povahu systému souborů HDFS s využitím dat uložených v účtu úložiště s povolenou službou Data Lake Storage Gen2.

Vytváření dotazů na data

Teď můžete začít vytvářet dotazy na data, která jste nahráli do svého účtu úložiště. Do buňky Cmd 1 zadejte všechny následující bloky kódu a stiskněte Cmd+Enter, abyste spustili skript Pythonu.

Chcete-li vytvořit datové rámce pro zdroje dat, spusťte následující skript:

  • Nahraďte <csv-folder-path> hodnotu zástupného symbolu cestou k souboru .csv .
# 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)

Tento skript zadejte, pokud chcete spustit některé základní dotazy analýzy pro data.

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

Vyčištění prostředků

Pokud už je nepotřebujete, odstraňte skupinu prostředků a všechny související prostředky. Provedete to tak, že vyberete skupinu prostředků pro účet úložiště a vyberete Odstranit.

Další kroky

Extrakce, transformace a načítání dat pomocí Apache Hivu ve službě Azure HDInsight