教學課程：使用 Apache Spark DataFrames 載入並轉換資料

本教學課程介紹如何使用 Azure Databricks 的 Apache Spark Python (PySpark) DataFrame API、Apache Spark Scala DataFrame API 及 SparkR SparkDataFrame API 載入並轉換資料。

在本教學課程結束時，您將了解什麼是 DataFrame 並熟悉以下工作：

Python（程式語言）

• 定義變數並將公用數據複製到 Unity 目錄磁碟區
• 使用 Python 建立 DataFrame
• 將資料從 CSV 檔案載入至 DataFrame
• 檢視 DataFrame 並與其互動
• 儲存 DataFrame
• 在 PySpark 執行 SQL 查詢

另請參閱Apache Spark PySpark API 參照。

程式語言 Scala

• 定義變數並將公用數據複製到 Unity 目錄磁碟區
• 使用 Scala 建立 DataFrame
• 將資料從 CSV 檔案載入至 DataFrame
• 檢視 DataFrame 並與其互動
• 儲存 DataFrame
• 在 Apache Spark 執行 SQL 查詢

另請參閱Apache Spark Scala API 參照.

R

• 定義變數並將公用數據複製到 Unity 目錄磁碟區
• 建立 SparkR SparkDataFrames
• 將資料從 CSV 檔案載入至 DataFrame
• 檢視 DataFrame 並與其互動
• 儲存 DataFrame
• 在 SparkR 執行 SQL 查詢

另請參閱Apache SparkR API 參照.

什麼是 DataFrame？

DataFrame 是具有可能不同類型之數據行的二維標籤數據結構。 您可以將 DataFrame 想像成電子表格、SQL 資料表或數位物件的字典。 Apache Spark DataFrame 提供一組豐富的函式（選取數據行、篩選、聯結、匯總），可讓您有效率地解決常見的數據分析問題。

Apache Spark DataFrame 是以彈性分散式資料集 (RDD) 為基礎建置的抽象概念。 Spark DataFrame 和 Spark SQL 使用統一的規劃和優化引擎，可讓您在 Azure Databricks （Python、SQL、Scala 和 R） 上取得所有支援語言的幾乎完全相同的效能。

需求

要完成以下教學課程，您必須滿足以下要求：

• 若要使用本教學課程中的範例，您的工作區必須已啟用 Unity 目錄。

• 本教學課程中的範例會使用 Unity 目錄 磁碟區 來儲存範例數據。 若要使用這些範例，請建立磁碟區，並使用該磁碟區的目錄、架構和磁碟區名稱來設定範例所使用的磁碟區路徑。

• 您必須在 Unity 目錄中具有下列權限：

  • READ VOLUME和WRITE VOLUME、或ALL PRIVILEGES本教學課程所使用的磁碟區。
  • 本教學課程所用架構的 USE SCHEMA 或 ALL PRIVILEGES。
  • 本教程使用的目錄為 USE CATALOG 或 ALL PRIVILEGES。

  若要設定這些權限，請聯絡 Databricks 管理員或 Unity Catalog 權限與可保護物件。

提示

有關本文的完整筆記本，請參閱DataFrame 教學課程筆記本。

步驟 1：定義變數並載入 CSV 檔案

此步驟會定義用於本教學課程的變數，然後載入包含嬰兒名字資料的 CSV 檔案，這些資料是從 health.data.ny.gov 匯入到您的 Unity Catalog 磁碟區中。

1. 按一下圖示以開啟新筆記本。 若要瞭解如何流覽 Azure Databricks 筆記本，請參閱 自定義筆記本外觀。

2. 請將下列程式碼複製並貼到全新空白筆記本資料格。 將 <catalog-name>、<schema-name>和 <volume-name> 替換為 Unity Catalog 的目錄名稱、架構名稱和磁碟區名稱。 以您選擇的數據表名稱取代 <table_name>。 稍後在本教學課程中，您會將嬰兒名稱數據載入此數據表。

   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_table = catalog + "." + schema
   print(path_table) # 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 downloadUrl = "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
   val fileName = "rows.csv"
   val tableName = "<table_name>"
   val pathVolume = s"/Volumes/$catalog/$schema/$volume"
   val pathTable = s"$catalog.$schema"
   print(pathVolume) // Show the complete path
   print(pathTable) // 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_table <- paste(catalog, ".", schema, sep = "")
   print(path_volume) # Show the complete path
   print(path_table) # Show the complete path
   

3. 按 Shift+Enter 以執行資料格並建立新的空白資料格。

4. 請將下列程式碼複製並貼到全新空白筆記本資料格。 此程式代碼會使用 rows.csv 命令，將 檔案從 health.data.ny.gov 複製到 Unity 目錄磁碟區。

   Python（程式語言）

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

   程式語言 Scala

   dbutils.fs.cp(downloadUrl, s"$pathVolume/$fileName")
   

   R

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

5. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

步驟 2：建立 DataFrame

此步驟建立一個以測試資料命名df1的 DataFrame，然後顯示其內容。

1. 請將下列程式碼複製並貼到全新空白筆記本資料格。 此程式代碼會使用測試數據建立 DataFrame，然後顯示 DataFrame 的內容和架構。

   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 = as.integer(c(2021)),
     First_Name = c("test"),
     County = c("Albany"),
     Sex = c("M"),
     Count = as.integer(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.
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

步驟 3：將資料從 CSV 檔案載入至 DataFrame

此步驟會從您先前載入 Unity 目錄磁碟區的 CSV 檔案建立名為 df_csv 的數據框架。 請參閱spark.read.csv。

1. 請將下列程式碼複製並貼到全新空白筆記本資料格。 此程式碼將嬰兒名稱資料df_csv從 CSV 檔案載入至 DataFrame，然後展示 DataFrame 的內容。

   Python（程式語言）

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

   程式語言 Scala

   val dfCsv = spark.read
       .option("header", "true")
       .option("inferSchema", "true")
       .option("delimiter", ",")
       .csv(s"$pathVolume/$fileName")
   
   display(dfCsv)
   

   R

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

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

您可以從許多支援的檔案格式載入資料。

步驟 4：檢視 DataFrame 並與其互動

使用以下方法檢視您的寶寶姓名 DataFrames 並與其互動。

列印 DataFrame 架構

瞭解如何顯示 Apache Spark DataFrame 的架構。 Apache Spark 會使用架構 一詞 來參考 DataFrame 中數據行的名稱和數據類型。

注意

Azure Databricks 也會使用架構一詞來描述向目錄註冊的數據表集合。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式代碼會使用 .printSchema() 方法來顯示DataFrame的架構，以檢視兩個DataFrame的架構- 以準備將兩個DataFrame聯集。

   Python（程式語言）

   df_csv.printSchema()
   df1.printSchema()
   

   程式語言 Scala

   dfCsv.printSchema()
   df1.printSchema()
   

   R

   printSchema(df_csv)
   printSchema(df1)
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

在 DataFrame 中重新命名數據行

瞭解如何重新命名 DataFrame 中的數據行。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式代碼會重新命名 df1_csv DataFrame 中的數據行，以符合 df1 DataFrame 中的個別數據行。 此程式碼使用 Apache SparkwithColumnRenamed()方法。

   Python（程式語言）

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

   程式語言 Scala

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

   R

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

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

合併 DataFrames

了解如何建立新的 DataFrame，將一個 DataFrame 的行列新增至另一個 DataFrame。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式碼使用 Apache Sparkunion()方法將第一個 DataFrame 的內容df 與df_csv包含從 CSV 檔案載入的嬰兒姓名資料的 DataFrame 合併。

   Python（程式語言）

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

   程式語言 Scala

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

   R

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

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

篩選 DataFrame 的行列

使用 Apache Spark .filter() 或 .where() 方法來篩選數據列，探索數據集中最受歡迎的嬰兒名稱。 使用篩選來選取數據列子集，以在DataFrame中傳回或修改。 效能或語法沒有差異，如以下範例所示。

使用 .filter() 方法

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式碼使用 Apache Spark .filter()方法來顯示 DataFrame 計數超過 50 的那些行列。

   Python（程式語言）

   display(df.filter(df["Count"] > 50))
   

   程式語言 Scala

   display(df.filter(df("Count") > 50))
   

   R

   display(filteredDF <- filter(df, df$Count > 50))
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

使用 .where（） 方法

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式碼使用 Apache Spark .where()方法來顯示 DataFrame 計數超過 50 的那些行列。

   Python（程式語言）

   display(df.where(df["Count"] > 50))
   

   程式語言 Scala

   display(df.where(df("Count") > 50))
   

   R

   display(filtered_df <- where(df, df$Count > 50))
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

從 DataFrame 選取數據行，並依頻率排序

請了解如何使用 select() 方法來指定要從 DataFrame 中返回的嬰兒名字頻率欄位。 使用 Apache Spark orderby和desc函式來排序結果。

Apache Spark 的 pyspark.sql 模組提供對 SQL 函式的支援。 我們在本教學課程使用的這些函式包括 Apache Spark orderBy()、desc()、以及expr()函式。 您可以根據需要將這些函式匯入工作階段來啟用這些功能。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式碼匯入 desc() 函式，然後使用 Apache Spark select() 方法與 Apache Spark orderBy() 以及 desc() 函式，按遞減順序顯示最常見的名稱及其計數。

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

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

建立子集合 DataFrame

了解如何從現有 DataFrame 建立子集合 DataFrame。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式碼使用 Apache Spark filter方法建立新的 DataFrame，按年份、計數及性別限制資料。 它會使用 Apache Spark select() 方法來限制數據行。 它還使用 Apache Spark orderBy()與desc()函式按計數對新的 DataFrame 進行排序。

   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)
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

步驟 5：儲存 DataFrame

了解如何儲存 DataFrame。 您可以將 DataFrame 儲存至數據表，或將數據框架寫入檔案或多個檔案。

將 DataFrame 儲存至數據表

根據預設，Azure Databricks 會針對所有數據表使用 Delta Lake 格式。 若要儲存 DataFrame，您必須在目錄和架構上擁有 CREATE 資料表許可權。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式代碼會使用您在本教學課程開始時定義的變數，將 DataFrame 的內容儲存至資料表。

   Python（程式語言）

   df.write.mode("overwrite").saveAsTable(f"{path_table}.{table_name}")
   

   程式語言 Scala

   df.write.mode("overwrite").saveAsTable(s"$pathTable" + "." + s"$tableName")
   

   R

   saveAsTable(df, paste(path_table, ".", table_name), mode = "overwrite")
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

大多數 Apache Spark 應用程式都以分散式方式處理大型資料集。 Apache Spark 會寫出檔案目錄，而非單一檔案。 Delta Lake 可拆分 Parquet 資料夾與檔案。 許多資料系統可以讀取這些檔案目錄。 Azure Databricks 建議針對大多數應用程式使用檔案路徑的數據表。

將 DataFrame 儲存至 JSON 檔案

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式碼將 DataFrame 儲存至 JSON 檔案目錄。

   Python（程式語言）

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

   程式語言 Scala

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

   R

   write.df(df, path = "/tmp/json_data", source = "json", mode = "overwrite")
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

從 JSON 檔案讀取 DataFrame

了解如何使用 Apache Spark spark.read.format()方法將 JSON 資料從目錄讀取至 DataFrame。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式碼顯示您在先前範例儲存的 JSON 檔案。

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

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

其他工作：在 PySpark、Scala 及 R 中執行 SQL 查詢

Apache Spark DataFrames 提供以下選項將 SQL 與 PySpark、Scala 及 R 結合。您可以在為本教學課程建立的同一筆記本中執行以下程式碼。

將數據行指定為 SQL 查詢

了解如何使用 Apache Spark selectExpr()方法。 這是select()方法的變體，可接受 SQL 運算式並傳回更新的 DataFrame。 該方法讓您可使用 SQL 運算式，例如upper。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式代碼會使用 Apache Spark selectExpr() 方法和 SQL upper 表示式，將字串資料行轉換成大寫（並重新命名數據行）。

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

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

使用 expr() 來套用欄位的 SQL 語法

瞭解如何匯入並使用 Apache Spark expr() 函式，以在指定數據行的任何位置使用 SQL 語法。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式代碼會匯入 expr() 函式，然後使用Apache Spark expr() 函式和 SQL lower 表示式，將字串數據行轉換成小寫（並將數據行重新命名）。

   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
   

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

使用 spark.sql() 函數執行任意 SQL 查詢

瞭解如何使用 Apache Spark spark.sql() 函數來執行任意 SQL 查詢。

1. 請將下列程式碼複製並貼到您空白筆記本資料格。 此程式代碼會使用 Apache Spark spark.sql() 函式，使用 SQL 語法查詢 SQL 數據表。

   Python（程式語言）

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

   程式語言 Scala

   display(spark.sql(s"SELECT * FROM $pathTable.$tableName"))
   

   R

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

2. 按 Shift+Enter 執行此資料格，然後移至下一個資料格。

DataFrame 教學課程筆記本

下列筆記本包含本教學課程中的範例查詢。

Python（程式語言）

使用 Python 的 DataFrames 教學課程

拿筆記本

程式語言 Scala

使用 Scala 的 DataFrames 教學課程

拿筆記本

R

使用 R 的 DataFrames 教學課程

拿筆記本

其他資源

• 在 Azure Databricks 的 PySpark
• Apache Spark APIs 參考
• 在 PySpark 與 pandas DataFrame 之間轉換
• Spark 上的 Pandas API