教程:在數據庫上使用Apache Spark Scala DataFrames

什麼是數據幀?

DataFrame是一種二維標記數據結構，其中的列可能具有不同的類型。您可以將DataFrame看作電子表格、SQL表或係列對象字典。Apache Spark DataFrames提供了一組豐富的函數(選擇列、篩選、連接、聚合)，使您能夠有效地解決常見的數據分析問題。

Apache Spark DataFrames是建立在彈性分布式數據集(rdd)之上的抽象。Spark DataFrames和Spark SQL使用統一的規劃和優化引擎，允許您在Databricks上的所有支持語言(Python、SQL、Scala和R)上獲得幾乎相同的性能。

什麼是Spark數據集?

Apache火花數據集的API提供類型安全的麵向對象編程接口。DataFrame是未類型化對象的別名數據集(行)．

Databricks文檔對大多數技術參考和指南使用了術語DataFrame，因為該語言包括Python、Scala和rScala數據集聚合器示例筆記本．

用Scala創建一個數據幀

大多數Apache Spark查詢返回一個DataFrame。這包括從表讀取數據、從文件加載數據以及轉換數據的操作。

你也可以從一個類列表中創建一個DataFrame，如下例所示:

情況下類員工（id：Int，的名字：字符串）瓦爾df＝Seq（新員工（1，“伊利亞”)，新員工（2，“Teo”)，新員工（3.，“方”))。toDF

將表讀入數據幀

Databricks默認對所有表使用Delta Lake。你可以輕鬆地將表加載到數據幀中，如下例所示:

火花．讀．表格（“< catalog_name >, < schema_name >。< table_name >”）

將數據從文件加載到DataFrame中

您可以從多個受支持的站點加載數據文件格式．中可用的數據集/ databricks-datasets目錄，可從大多數工作空間訪問。看到樣本數據集．

瓦爾df＝火花．讀．格式（“csv”）．選項（“頭”，“真正的”）．選項（“inferSchema”，“真正的”）．負載（“/ databricks-datasets /樣本/ population-vs-price / data_geo.csv”）

將轉換步驟分配給DataFrame

大多數Spark轉換的結果返回一個DataFrame。您可以將這些結果分配回DataFrame變量，類似於您在其他係統中使用cte、臨時視圖或DataFrame的方式。

使用join和union組合數據幀

數據幀使用標準的SQL語義進行連接操作。連接根據提供的匹配條件和連接類型返回兩個數據幀的組合結果。下麵的例子是一個默認的內部連接:

瓦爾joined_df＝df1．加入（df2，joinType＝“內心”，usingColumn＝“id”）

你可以使用聯合操作將一個DataFrame的行添加到另一個DataFrame中，如下例所示:

瓦爾unioned_df＝df1．聯盟（df2）

篩選數據幀中的行

您可以使用.filter ()orgydF4y2Ba其中()．在性能和語法上沒有區別，如下例所示:

瓦爾filtered_df＝df．過濾器（“id > 1”）瓦爾filtered_df＝df．在哪裏（“id > 1”）

使用篩選來選擇DataFrame中要返回或修改的行子集。

從數據幀中選擇列

可以通過傳遞一個或多個列名來選擇列.select ()，如下例:

瓦爾select_df＝df．選擇（“id”，“名稱”）

您可以結合選擇和篩選查詢來限製返回的行和列。

subset_df＝df．過濾器（“id > 1”）.選擇（“名稱”）

查看DataFrame

要以表格格式查看此數據，可以使用Databricks顯示()命令，示例如下:

顯示（df）

打印數據模式

Spark使用了這個術語模式以引用DataFrame中列的名稱和數據類型。

方法可以打印模式.printSchema ()方法，如下例所示:

df．printSchema（）

將數據幀保存到表中

Databricks默認對所有表使用Delta Lake。你可以使用以下語法將DataFrame的內容保存到一個表中:

df．寫．toTable（“< table_name >”）

將數據幀寫入文件集合中

大多數Spark應用程序都被設計為處理大型數據集並以分布式方式工作，而且Spark會寫出一個目錄的文件，而不是單個文件。許多數據係統被配置為讀取這些文件目錄。Databricks建議對大多數應用程序使用表而不是文件路徑。

下麵的例子保存了一個JSON文件目錄:

df．寫．格式（“json”）.保存（“/ tmp / json_data”）

在Spark中運行SQL查詢

Spark DataFrames提供了許多將SQL與Scala結合起來的選項。

的selectExpr ()方法允許您將每一列指定為一個SQL查詢，如下例所示:

顯示（df．selectExpr（“id”，“上(名稱)作為big_name”）)

您可以導入expr ()函數pyspark.sql.functions在指定列的任何地方使用SQL語法，如下例所示:

進口org．apache．火花．sql．功能．expr顯示（df．選擇（的id，expr（“低(名稱)作為little_name”))）

你也可以用spark.sql ()在Scala內核中運行任意SQL查詢，如下例所示:

瓦爾query_df＝火花．sql（"SELECT * FROM "）

因為邏輯是在Scala內核中執行的，所有的SQL查詢都是作為字符串傳遞的，你可以使用Scala格式化來參數化SQL查詢，如下麵的例子所示:

瓦爾table_name＝“my_table”瓦爾query_df＝火花．sql（“SELECT *＄table_name＂）

Scala數據集聚合器示例筆記本

這裏有一個筆記本，向您展示如何使用數據集聚合器。