Självstudie: Läsa in och transformera data med Apache Spark DataFrames

  • Artikel

Den här självstudien visar hur du läser in och transformerar data med apache Spark Python (PySpark) DataFrame API, Apache Spark Scala DataFrame API och SparkR SparkDataFrame API i Azure Databricks.

I slutet av den här självstudien kommer du att förstå vad en DataFrame är och känna till följande uppgifter:

Python

Se även Apache Spark PySpark API-referens.

Scala

Se även Apache Spark Scala API-referens.

R

Se även Apache SparkR API-referens.

Vad är en DataFrame?

En DataFrame är en tvådimensionell etiketterad datastruktur med kolumner av potentiellt olika typer. Du kan tänka dig en DataFrame som ett kalkylblad, en SQL-tabell eller en ordlista med serieobjekt. Apache Spark DataFrames tillhandahåller en omfattande uppsättning funktioner (välj kolumner, filtrera, koppla, aggregera) som gör att du kan lösa vanliga dataanalysproblem effektivt.

Apache Spark DataFrames är en abstraktion som bygger på Resilient Distributed Datasets (RDD). Spark DataFrames och Spark SQL använder en enhetlig planerings- och optimeringsmotor så att du kan få nästan identiska prestanda för alla språk som stöds på Azure Databricks (Python, SQL, Scala och R).

Krav

För att slutföra följande självstudie måste du uppfylla följande krav:

  • Om du vill använda exemplen i den här självstudien måste din arbetsyta ha Unity Catalog aktiverat.

  • Exemplen i den här självstudien använder en Unity Catalog-volym för att lagra exempeldata. Om du vill använda dessa exempel skapar du en volym och använder volymens katalog-, schema- och volymnamn för att ange den volymsökväg som används av exemplen.

  • Du måste ha följande behörigheter i Unity Catalog:

    • READ VOLUME och WRITE VOLUME, eller ALL PRIVILEGES för volymen som används för den här självstudien.
    • USE SCHEMA eller ALL PRIVILEGES för schemat som används för den här självstudien.
    • USE CATALOG eller ALL PRIVILEGES för katalogen som används för den här självstudien.

    Information om hur du anger dessa behörigheter finns i Behörigheter för Databricks-administratören eller Unity Catalog och skyddsbara objekt.

Steg 1: Definiera variabler och läsa in CSV-fil

Det här steget definierar variabler för användning i den här självstudien och läser sedan in en CSV-fil som innehåller babynamnsdata från health.data.ny.gov till unity-katalogvolymen.

  1. Öppna en ny anteckningsbok genom att Ny ikon klicka på ikonen. Information om hur du navigerar i Notebook-filer för Azure Databricks finns i Databricks Notebook-gränssnitt och -kontroller.

  2. Kopiera och klistra in följande kod i den nya tomma notebook-cellen. Ersätt <catalog-name>, <schema-name>och <volume-name> med katalog-, schema- och volymnamnen för en Unity Catalog-volym. Ersätt <table_name> med ett valfritt tabellnamn. Du läser in babynamnsdata i den här tabellen senare i den här självstudien.

  3. Tryck Shift+Enter för att köra cellen och skapa en ny tom cell.

    Python

    catalog = "<catalog_name>"
schema = "<schema_name>"
volume = "<volume_name>"
download_url = "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
file_name = "rows.csv"
table_name = "<table_name>"
path_volume = "/Volumes/" + catalog + "/" + schema + "/" + volume
path_tables = catalog + "." + schema
print(path_tables) # Show the complete path
print(path_volume) # Show the complete path

    Scala

    val catalog = "<catalog_name>"
val schema = "<schema_name>"
val volume = "<volume_name>"
val download_url = "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
val file_name = "rows.csv"
val table_name = "<table_name>"
val path_volume = s"/Volumes/$catalog/$schema/$volume"
val path_tables = s"$catalog.$schema.$table_name"
print(path_volume) // Show the complete path
print(path_tables) // Show the complete path

    R

    catalog <- "<catalog_name>"
schema <- "<schema_name>"
volume <- "<volume_name>"
download_url <- "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
file_name <- "rows.csv"
table_name <- "<table_name>"
path_volume <- paste("/Volumes/", catalog, "/", schema, "/", volume, sep = "")
path_tables <- paste(catalog, ".", schema, sep = "")
print(path_volume) # Show the complete path
print(path_tables) # Show the complete path

  4. Kopiera och klistra in följande kod i den nya tomma notebook-cellen. Den här koden kopierar rows.csv filen från health.data.ny.gov till Unity Catalog-volymen med hjälp av kommandot Databricks dbutuils .

  5. Tryck Shift+Enter för att köra cellen och flytta sedan till nästa cell.

    Python

    dbutils.fs.cp(f"{download_url}", f"{path_volume}" + "/" + f"{file_name}")

    Scala

    dbutils.fs.cp(download_url, s"$path_volume/$file_name")

    R

    dbutils.fs.cp(download_url, paste(path_volume, "/", file_name, sep = ""))

Steg 2: Skapa en dataram

Det här steget skapar en DataFrame med namnet df1 med testdata och visar sedan dess innehåll.

  1. Kopiera och klistra in följande kod i den nya tomma notebook-cellen. Den här koden skapar dataramen med testdata och visar sedan innehållet och schemat för DataFrame.

  2. Tryck Shift+Enter för att köra cellen och flytta sedan till nästa cell.

    Python

    data = [[2021, "test", "Albany", "M", 42]]
columns = ["Year", "First_Name", "County", "Sex", "Count"]

df1 = spark.createDataFrame(data, schema="Year int, First_Name STRING, County STRING, Sex STRING, Count int")
display(df1) # The display() method is specific to Databricks notebooks and provides a richer visualization.
# df1.show() The show() method is a part of the Apache Spark DataFrame API and provides basic visualization.

    Scala

    val data = Seq((2021, "test", "Albany", "M", 42))
val columns = Seq("Year", "First_Name", "County", "Sex", "Count")

val df1 = data.toDF(columns: _*)
display(df1) // The display() method is specific to Databricks notebooks and provides a richer visualization.
// df1.show() The show() method is a part of the Apache Spark DataFrame API and provides basic visualization.

    R

    # Load the SparkR package that is already preinstalled on the cluster.
library(SparkR)

data <- data.frame(
  Year = c(2021),
  First_Name = c("test"),
  County = c("Albany"),
  Sex = c("M"),
  Count = c(42)
)

df1 <- createDataFrame(data)
display(df1) # The display() method is specific to Databricks notebooks and provides a richer visualization.
# head(df1) The head() method is a part of the Apache SparkR DataFrame API and provides basic visualization.

Steg 3: Läsa in data i en DataFrame från CSV-fil

Det här steget skapar en DataFrame med namnet df_csv från CSV-filen som du tidigare läste in i Unity Catalog-volymen. Se spark.read.csv.

  1. Kopiera och klistra in följande kod i den nya tomma notebook-cellen. Den här koden läser in babynamnsdata i DataFrame df_csv från CSV-filen och visar sedan innehållet i DataFrame.

  2. Tryck Shift+Enter för att köra cellen och flytta sedan till nästa cell.

    Python

    df_csv = spark.read.csv(f"{path_volume}/{file_name}",
  header=True,
  inferSchema=True,
  sep=",")
display(df_csv)

    Scala

    val df_csv = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .option("delimiter", ",")
  .csv(s"$path_volume/$file_name")

display(df_csv)

    R

    df_csv <- read.df(paste(path_volume, "/", file_name, sep=""),
  source="csv",
  header = TRUE,
  inferSchema = TRUE,
  delimiter = ",")

display(df_csv)

Du kan läsa in data från många filformat som stöds.

Steg 4: Visa och interagera med din DataFrame

Visa och interagera med dina babynamn DataFrames med hjälp av följande metoder.

Lär dig hur du visar schemat för en Apache Spark DataFrame. Apache Spark använder termschemat för att referera till namnen och datatyperna för kolumnerna i DataFrame.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden visar schemat för dina DataFrames med .printSchema() metoden för att visa scheman för de två DataFrames – för att förbereda för att unionera de två DataFrames.

Python

df_csv.printSchema()
df1.printSchema()

Scala

df_csv.printSchema()
df1.printSchema()

R

printSchema(df_csv)
printSchema(df1)

Kommentar

Azure Databricks använder också termschemat för att beskriva en samling tabeller som är registrerade i en katalog.

Byt namn på kolumnen i DataFrame

Lär dig hur du byter namn på en kolumn i en DataFrame.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden byter namn på en kolumn i df1_csv DataFrame så att den matchar respektive kolumn i df1 DataFrame. Den här koden använder Apache Spark-metoden withColumnRenamed() .

Python

df_csv = df_csv.withColumnRenamed("First Name", "First_Name")
df_csv.printSchema

Scala

val df_csvRenamed = df_csv.withColumnRenamed("First Name", "First_Name")
// when modifying a DataFrame in Scala, you must assign it to a new variable
df_csv_renamed.printSchema()

R

df_csv <- withColumnRenamed(df_csv, "First Name", "First_Name")
printSchema(df_csv)

Kombinera DataFrames

Lär dig hur du skapar en ny DataFrame som lägger till raderna i en DataFrame till en annan.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden använder Apache Spark-metoden union() för att kombinera innehållet i din första DataFrame df med DataFrame df_csv som innehåller babynamndata som lästs in från CSV-filen.

Python

df = df1.union(df_csv)
display(df)

Scala

val df = df1.union(df_csv_renamed)
display(df)

R

display(df <- union(df1, df_csv))

Filtrera rader i en dataram

Identifiera de mest populära babynamnen i datauppsättningen genom att filtrera rader med hjälp av Apache Spark .filter() eller .where() metoder. Använd filtrering för att välja en delmängd rader som ska returneras eller ändras i en DataFrame. Det finns ingen skillnad i prestanda eller syntax, som du ser i följande exempel.

Använda metoden .filter()

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden använder Apache Spark-metoden .filter() för att visa dessa rader i DataFrame med ett antal på mer än 50.

Python
display(df.filter(df["Count"] > 50))
Scala
display(df.filter(df("Count") > 50))
R
display(filteredDF <- filter(df, df$Count > 50))

Använda .where()-metod

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden använder Apache Spark-metoden .where() för att visa dessa rader i DataFrame med ett antal på mer än 50.

Python
display(df.where(df["Count"] > 50))
Scala
display(df.where(df("Count") > 50))
R
display(filtered_df <- where(df, df$Count > 50))

Välj kolumner från en DataFrame och sortera efter frekvens

Lär dig mer om vilken babynamnsfrekvens med select() metoden för att ange kolumnerna från DataFrame som ska returneras. Använd Apache Spark orderby och desc funktioner för att ordna resultatet.

Modulen pyspark.sql för Apache Spark har stöd för SQL-funktioner. Bland de här funktionerna som vi använder i den här självstudien finns funktionerna Apache Spark orderBy(), desc()och expr() . Du aktiverar användningen av dessa funktioner genom att importera dem till sessionen efter behov.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden importerar desc() funktionen och använder sedan Apache Spark-metoden select() och Apache Spark orderBy() och desc() funktioner för att visa de vanligaste namnen och deras antal i fallande ordning.

Python

from pyspark.sql.functions import desc
display(df.select("First_Name", "Count").orderBy(desc("Count")))

Scala

import org.apache.spark.sql.functions.desc
display(df.select("First_Name", "Count").orderBy(desc("Count")))

R

display(arrange(select(df, df$First_Name, df$Count), desc(df$Count)))

Skapa en delmängd av DataFrame

Lär dig hur du skapar en delmängd av DataFrame från en befintlig DataFrame.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden använder Apache Spark-metoden filter för att skapa en ny DataFrame som begränsar data efter år, antal och kön. Den använder Apache Spark-metoden select() för att begränsa kolumnerna. Den använder också Apache Spark orderBy() och desc() funktioner för att sortera den nya DataFrame efter antal.

Python

subsetDF = df.filter((df["Year"] == 2009) & (df["Count"] > 100) & (df["Sex"] == "F")).select("First_Name", "County", "Count").orderBy(desc("Count"))
display(subsetDF)

Scala

val subsetDF = df.filter((df("Year") == 2009) && (df("Count") > 100) && (df("Sex") == "F")).select("First_Name", "County", "Count").orderBy(desc("Count"))

display(subsetDF)

R

subsetDF <- select(filter(df, (df$Count > 100) & (df$year == 2009) & df["Sex"] == "F")), "First_Name", "County", "Count")
display(subsetDF)

Steg 5: Spara dataramen

Lär dig hur du sparar en DataFrame. Du kan antingen spara dataramen i en tabell eller skriva DataFrame till en fil eller flera filer.

Spara DataFrame i en tabell

Azure Databricks använder Delta Lake-formatet för alla tabeller som standard. Om du vill spara dataramen måste du ha CREATE tabellbehörigheter i katalogen och schemat.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden sparar innehållet i DataFrame till en tabell med hjälp av variabeln som du definierade i början av den här självstudien.

Python

df.write.saveAsTable(s"$path_tables" + "." + s"$table_name")

# To overwrite an existing table, use the following code:
# df.write.mode("overwrite").saveAsTable(fs"$path_tables" + "." + s"$table_name")

Scala

df.write.saveAsTable(s"$path_tables" + "." + s"$table_name")

// To overwrite an existing table, use the following code:
// df.write.mode("overwrite").saveAsTable(s"$path_volume" + "." + s"$table_name")

R

saveAsTable(df, paste(path_tables, ".", table_name))
# To overwrite an existing table, use the following code:
# saveAsTable(df, paste(path_tables, ".", table_name), mode = "overwrite")

De flesta Apache Spark-program fungerar på stora datamängder och på ett distribuerat sätt. Apache Spark skriver ut en katalog med filer i stället för en enda fil. Delta Lake delar upp Parquet-mapparna och filerna. Många datasystem kan läsa dessa kataloger med filer. Azure Databricks rekommenderar att du använder tabeller över filsökvägar för de flesta program.

Spara DataFrame till JSON-filer

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden sparar DataFrame till en katalog med JSON-filer.

Python

df.write.format("json").save("/tmp/json_data")

# To overwrite an existing file, use the following code:
# df.write.format("json").mode("overwrite").save("/tmp/json_data")

Scala

df.write.format("json").save("/tmp/json_data")

// To overwrite an existing file, use the following code:
// df.write.format("json").mode("overwrite").save("/tmp/json_data")

R

write.df(df, path = "/tmp/json_data", source = "json")
# To overwrite an existing file, use the following code:
# write.df(df, path = "/tmp/json_data", source = "json", mode = "overwrite")

Läsa DataFrame från en JSON-fil

Lär dig hur du använder Apache Spark-metoden spark.read.format() för att läsa JSON-data från en katalog till en DataFrame.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden visar JSON-filerna som du sparade i föregående exempel.

Python

display(spark.read.format("json").json("/tmp/json_data"))

Scala

display(spark.read.format("json").json("/tmp/json_data"))

R

display(read.json("/tmp/json_data"))

Ytterligare uppgifter: Köra SQL-frågor i PySpark, Scala och R

Apache Spark DataFrames innehåller följande alternativ för att kombinera SQL med PySpark, Scala och R. Du kan köra följande kod i samma notebook-fil som du skapade för den här självstudien.

Ange en kolumn som en SQL-fråga

Lär dig hur du använder Apache Spark-metoden selectExpr() . Det här är en variant av metoden select() som accepterar SQL-uttryck och returnerar en uppdaterad DataFrame. Med den här metoden kan du använda ett SQL-uttryck, till exempel upper.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden använder Apache Spark-metoden selectExpr() och SQL-uttrycket upper för att konvertera en strängkolumn till versaler (och byta namn på kolumnen).

Python

display(df.selectExpr("Count", "upper(County) as big_name"))

Scala

display(df.selectExpr("Count", "upper(County) as big_name"))

R

display(df_selected <- selectExpr(df, "Count", "upper(County) as big_name"))

Använd expr() för att använda SQL-syntax för en kolumn

Lär dig hur du importerar och använder Apache Spark-funktionen expr() för att använda SQL-syntax var som helst där en kolumn skulle anges.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden importerar expr() funktionen och använder sedan Apache Spark-funktionen expr() och SQL-uttrycket lower för att konvertera en strängkolumn till gemener (och byta namn på kolumnen).

Python

from pyspark.sql.functions import expr
display(df.select("Count", expr("lower(County) as little_name")))

Scala

import org.apache.spark.sql.functions.{col, expr} // Scala requires us to import the col() function as well as the expr() function

display(df.select(col("Count"), expr("lower(County) as little_name")))

R

display(df_selected <- selectExpr(df, "Count", "lower(County) as little_name"))
# expr() function is not supported in R, selectExpr in SparkR replicates this functionality

Köra en godtycklig SQL-fråga med hjälp av funktionen spark.sql()

Lär dig hur du använder Apache Spark-funktionen spark.sql() för att köra godtyckliga SQL-frågor.

Kopiera och klistra in följande kod i en tom notebook-cell. Den här koden använder Apache Spark-funktionen spark.sql() för att köra frågor mot en SQL-tabell med sql-syntax.

Python

display(spark.sql(f"SELECT * FROM {path_tables}" + "." + f"{table_name}"))

Scala

display(spark.sql(s"SELECT * FROM $path_tables.$table_name"))

R

display(sql(paste("SELECT * FROM", path_tables, ".", table_name)))

Notebook-fil för DataFrame-självstudie

Följande notebook-fil innehåller exempelfrågor från den här självstudien.

Python

Självstudie om DataFrames-notebook-fil

Hämta notebook-fil

Scala

Självstudie om DataFrames-notebook-fil

Hämta notebook-fil

R

Självstudie om DataFrames-notebook-fil

Hämta notebook-fil

Ytterligare resurser