在 R 中使用 DataFrame 和資料表
本文說明如何使用 SparkR 、sparklyr 和 dplyr 等 R 套件來處理 R data.frame
s、 Spark DataFrame 和 記憶體內部 資料表。
請注意,當您使用 SparkR、sparklyr 和 dplyr 時,您可能會發現您可以使用所有這些套件完成特定作業,而且您可以使用最熟悉的套件。 例如,若要執行查詢,您可以呼叫 、 sparklyr::sdf_sql
、 和 dplyr::select
等 SparkR::sql
函式。 在其他時候,您可能只要完成一或兩個這些套件的作業,而您選擇的作業取決於您的使用案例。 例如,您呼叫 sparklyr::sdf_quantile
的方式與呼叫 dplyr::percentile_approx
的方式稍有不同,即使這兩個函式都會計算分位數。
您可以使用 SQL 作為 SparkR 與 sparklyr 之間的橋接器。 例如,您可以使用 SparkR::sql
來查詢使用 sparklyr 建立的資料表。 您可以使用 sparklyr::sdf_sql
來查詢使用 SparkR 建立的資料表。 程式 dplyr
代碼在執行之前,一律會轉譯為記憶體中的 SQL。 另 請參閱 API 互通性 和 SQL 翻譯 。
SparkR、sparklyr 和 dplyr 套件包含在 Azure Databricks 叢集 上安裝的 Databricks Runtime 中。 因此,您不需要呼叫一般 install.package
,才能開始呼叫這些套件。 不過,您仍必須先載入 library
這些套件。 例如,從 Azure Databricks 工作區的 R 筆記本 內,在 Notebook 資料格中執行下列程式碼以載入 SparkR、sparklyr 和 dplyr:
library(SparkR)
library(sparklyr)
library(dplyr)
載入 sparklyr 之後,您必須呼叫 sparklyr::spark_connect
以連線到叢集,並指定 databricks
連接方法。 例如,在筆記本資料格中執行下列程式碼,以連線到裝載筆記本的叢集:
sc <- spark_connect(method = "databricks")
相反地,Azure Databricks 筆記本已在叢集上建立 SparkSession
,以便與 SparkR 搭配使用,因此您不需要呼叫 SparkR::sparkR.session
,才能開始呼叫 SparkR。
本文中的許多程式碼範例都是以 Azure Databricks 工作區中特定位置的資料為基礎,其中包含特定資料行名稱和資料類型。 此程式碼範例的資料來源自于 GitHub 內名為 的 book.json
JSON 檔案。 若要取得此檔案,並將它上傳至您的工作區:
- 移至 GitHub 上的 books.json 檔案,並使用文字編輯器將其內容複寫到本機電腦上某處名為
books.json
的檔案。 - 在 Azure Databricks 工作區提要欄位中,按一下 [目錄 ]。
- 按一下 [ 建立資料表 ]。
- 在 [ 上傳檔案 ] 索引標籤上
books.json
,將檔案從本機電腦卸載至 [要上傳的 檔案] 方塊。 或選取 按一下以流覽 ,然後從本機電腦流覽至books.json
檔案。
根據預設,Azure Databricks 會使用 路徑 /FileStore/tables/books.json
,將本機 books.json
檔案上傳至 工作區中的 DBFS 位置。
請勿按一下 [使用 UI 建立資料表] 或 [ 在筆記本 中建立資料表]。 本文中的程式碼範例會使用此 DBFS 位置中上傳 books.json
檔案中的資料。
使用 sparklyr::spark_read_json
將上傳的 JSON 檔案讀取至 DataFrame、指定連接、JSON 檔案的路徑,以及資料內部資料表表示的名稱。 在此範例中,您必須指定 book.json
檔案包含多行。 在這裡指定資料行的架構是選擇性的。 否則,sparklyr 預設會推斷資料行的架構。 例如,在筆記本資料格中執行下列程式碼,將上傳的 JSON 檔案資料讀入名為 jsonDF
的資料框架中:
jsonDF <- spark_read_json(
sc = sc,
name = "jsonTable",
path = "/FileStore/tables/books.json",
options = list("multiLine" = TRUE),
columns = c(
author = "character",
country = "character",
imageLink = "character",
language = "character",
link = "character",
pages = "integer",
title = "character",
year = "integer"
)
)
您可以使用 SparkR::head
、 SparkR::show
或 sparklyr::collect
來列印 DataFrame 的第一個資料列。 根據預設, head
預設會列印前六個數據列。 show
並 collect
列印前 10 個數據列。 例如,在筆記本資料格中執行下列程式碼,以列印名為 jsonDF
之 DataFrame 的第一個資料列:
head(jsonDF)
# Source: spark<?> [?? x 8]
# author country image…¹ langu…² link pages title year
# <chr> <chr> <chr> <chr> <chr> <int> <chr> <int>
# 1 Chinua Achebe Nigeria images… English "htt… 209 Thin… 1958
# 2 Hans Christian Andersen Denmark images… Danish "htt… 784 Fair… 1836
# 3 Dante Alighieri Italy images… Italian "htt… 928 The … 1315
# 4 Unknown Sumer and Akk… images… Akkadi… "htt… 160 The … -1700
# 5 Unknown Achaemenid Em… images… Hebrew "htt… 176 The … -600
# 6 Unknown India/Iran/Ir… images… Arabic "htt… 288 One … 1200
# … with abbreviated variable names ¹imageLink, ²language
show(jsonDF)
# Source: spark<jsonTable> [?? x 8]
# author country image…¹ langu…² link pages title year
# <chr> <chr> <chr> <chr> <chr> <int> <chr> <int>
# 1 Chinua Achebe Nigeria images… English "htt… 209 Thin… 1958
# 2 Hans Christian Andersen Denmark images… Danish "htt… 784 Fair… 1836
# 3 Dante Alighieri Italy images… Italian "htt… 928 The … 1315
# 4 Unknown Sumer and Ak… images… Akkadi… "htt… 160 The … -1700
# 5 Unknown Achaemenid E… images… Hebrew "htt… 176 The … -600
# 6 Unknown India/Iran/I… images… Arabic "htt… 288 One … 1200
# 7 Unknown Iceland images… Old No… "htt… 384 Njál… 1350
# 8 Jane Austen United Kingd… images… English "htt… 226 Prid… 1813
# 9 Honoré de Balzac France images… French "htt… 443 Le P… 1835
# 10 Samuel Beckett Republic of … images… French… "htt… 256 Moll… 1952
# … with more rows, and abbreviated variable names ¹imageLink, ²language
# ℹ Use `print(n = ...)` to see more rows
collect(jsonDF)
# A tibble: 100 × 8
# author country image…¹ langu…² link pages title year
# <chr> <chr> <chr> <chr> <chr> <int> <chr> <int>
# 1 Chinua Achebe Nigeria images… English "htt… 209 Thin… 1958
# 2 Hans Christian Andersen Denmark images… Danish "htt… 784 Fair… 1836
# 3 Dante Alighieri Italy images… Italian "htt… 928 The … 1315
# 4 Unknown Sumer and Ak… images… Akkadi… "htt… 160 The … -1700
# 5 Unknown Achaemenid E… images… Hebrew "htt… 176 The … -600
# 6 Unknown India/Iran/I… images… Arabic "htt… 288 One … 1200
# 7 Unknown Iceland images… Old No… "htt… 384 Njál… 1350
# 8 Jane Austen United Kingd… images… English "htt… 226 Prid… 1813
# 9 Honoré de Balzac France images… French "htt… 443 Le P… 1835
# 10 Samuel Beckett Republic of … images… French… "htt… 256 Moll… 1952
# … with 90 more rows, and abbreviated variable names ¹imageLink, ²language
# ℹ Use `print(n = ...)` to see more rows
您可以使用 dplyr 函式在 DataFrame 上執行 SQL 查詢。 例如,在筆記本資料格中執行下列程式碼,以使用 dplyr::group_by
,並從 dployr::count
名為 jsonDF
的 DataFrame 取得計數。 使用 dplyr::arrange
和 dplyr::desc
依計數以遞減順序排序結果。 然後預設會列印前 10 個數據列。
group_by(jsonDF, author) %>%
count() %>%
arrange(desc(n))
# Source: spark<?> [?? x 2]
# Ordered by: desc(n)
# author n
# <chr> <dbl>
# 1 Fyodor Dostoevsky 4
# 2 Unknown 4
# 3 Leo Tolstoy 3
# 4 Franz Kafka 3
# 5 William Shakespeare 3
# 6 William Faulkner 2
# 7 Gustave Flaubert 2
# 8 Homer 2
# 9 Gabriel García Márquez 2
# 10 Thomas Mann 2
# … with more rows
# ℹ Use `print(n = ...)` to see more rows
然後 sparklyr::spark_write_table
,您可以使用 將結果寫入 Azure Databricks 中的資料表。 例如,在筆記本資料格中執行下列程式碼以重新執行查詢,然後將結果寫入名為 json_books_agg
的資料表:
group_by(jsonDF, author) %>%
count() %>%
arrange(desc(n)) %>%
spark_write_table(
name = "json_books_agg",
mode = "overwrite"
)
若要確認資料表已建立,您可以搭配 SparkR::showDF
使用 sparklyr::sdf_sql
來顯示資料表的資料。 例如,在筆記本資料格中執行下列程式碼,以查詢資料表到 DataFrame,然後使用 sparklyr::collect
預設列印 DataFrame 的前 10 個數據列:
collect(sdf_sql(sc, "SELECT * FROM json_books_agg"))
# A tibble: 82 × 2
# author n
# <chr> <dbl>
# 1 Fyodor Dostoevsky 4
# 2 Unknown 4
# 3 Leo Tolstoy 3
# 4 Franz Kafka 3
# 5 William Shakespeare 3
# 6 William Faulkner 2
# 7 Homer 2
# 8 Gustave Flaubert 2
# 9 Gabriel García Márquez 2
# 10 Thomas Mann 2
# … with 72 more rows
# ℹ Use `print(n = ...)` to see more rows
您也可以使用 sparklyr::spark_read_table
來執行類似的事情。 例如,在筆記本資料格中執行下列程式碼,以查詢名為 jsonDF
的先前 DataFrame,然後依預設使用 sparklyr::collect
來列印 DataFrame 的前 10 個數據列:
fromTable <- spark_read_table(
sc = sc,
name = "json_books_agg"
)
collect(fromTable)
# A tibble: 82 × 2
# author n
# <chr> <dbl>
# 1 Fyodor Dostoevsky 4
# 2 Unknown 4
# 3 Leo Tolstoy 3
# 4 Franz Kafka 3
# 5 William Shakespeare 3
# 6 William Faulkner 2
# 7 Homer 2
# 8 Gustave Flaubert 2
# 9 Gabriel García Márquez 2
# 10 Thomas Mann 2
# … with 72 more rows
# ℹ Use `print(n = ...)` to see more rows
您可以使用 dplyr 函式,將資料行新增至 DataFrame,以及計算資料行的值。
例如,在筆記本資料格中執行下列程式碼,以取得名為 jsonDF
的資料框架內容。 使用 dplyr::mutate
新增名為 today
的資料行,並將這個新資料行填入目前的時間戳記。 然後將這些內容寫入名為 withDate
的新 DataFrame,並使用 dplyr::collect
預設列印新 DataFrame 的前 10 個數據列。
備註
dplyr::mutate
只接受符合 Hive 內建函式 (也稱為 UDF) 和內建彙總函式 (也稱為 UDAF) 的引數。 如需一般資訊,請參閱 Hive 函式 。 如需本節中日期相關函式的相關資訊,請參閱 Date Functions 。
withDate <- jsonDF %>%
mutate(today = current_timestamp())
collect(withDate)
# A tibble: 100 × 9
# author country image…¹ langu…² link pages title year today
# <chr> <chr> <chr> <chr> <chr> <int> <chr> <int> <dttm>
# 1 Chinua A… Nigeria images… English "htt… 209 Thin… 1958 2022-09-27 21:32:59
# 2 Hans Chr… Denmark images… Danish "htt… 784 Fair… 1836 2022-09-27 21:32:59
# 3 Dante Al… Italy images… Italian "htt… 928 The … 1315 2022-09-27 21:32:59
# 4 Unknown Sumer … images… Akkadi… "htt… 160 The … -1700 2022-09-27 21:32:59
# 5 Unknown Achaem… images… Hebrew "htt… 176 The … -600 2022-09-27 21:32:59
# 6 Unknown India/… images… Arabic "htt… 288 One … 1200 2022-09-27 21:32:59
# 7 Unknown Iceland images… Old No… "htt… 384 Njál… 1350 2022-09-27 21:32:59
# 8 Jane Aus… United… images… English "htt… 226 Prid… 1813 2022-09-27 21:32:59
# 9 Honoré d… France images… French "htt… 443 Le P… 1835 2022-09-27 21:32:59
# 10 Samuel B… Republ… images… French… "htt… 256 Moll… 1952 2022-09-27 21:32:59
# … with 90 more rows, and abbreviated variable names ¹imageLink, ²language
# ℹ Use `print(n = ...)` to see more rows
現在,使用 dplyr::mutate
將兩個數據行新增至 DataFrame 的內容 withDate
。 新的 month
和 year
資料行包含資料行中的 today
數值月份和年份。 然後將這些內容寫入名為 withMMyyyy
的新 DataFrame,並搭配 dplyr::collect
使用 dplyr::select
,並依預設列印 author
新 DataFrame 前十個數據列的 、 title
month
和資料 year
行:
withMMyyyy <- withDate %>%
mutate(month = month(today),
year = year(today))
collect(select(withMMyyyy, c("author", "title", "month", "year")))
# A tibble: 100 × 4
# author title month year
# <chr> <chr> <int> <int>
# 1 Chinua Achebe Things Fall Apart 9 2022
# 2 Hans Christian Andersen Fairy tales 9 2022
# 3 Dante Alighieri The Divine Comedy 9 2022
# 4 Unknown The Epic Of Gilgamesh 9 2022
# 5 Unknown The Book Of Job 9 2022
# 6 Unknown One Thousand and One Nights 9 2022
# 7 Unknown Njál's Saga 9 2022
# 8 Jane Austen Pride and Prejudice 9 2022
# 9 Honoré de Balzac Le Père Goriot 9 2022
# 10 Samuel Beckett Molloy, Malone Dies, The Unnamable, the … 9 2022
# … with 90 more rows
# ℹ Use `print(n = ...)` to see more rows
現在,使用 dplyr::mutate
將兩個數據行新增至 DataFrame 的內容 withMMyyyy
。 新 formatted_date
資料行包含 yyyy-MM-dd
資料行中的 today
部分,而新 day
資料行則包含新 formatted_date
資料行中的數值日。 然後將這些內容寫入名為 withUnixTimestamp
的新 DataFrame,並搭配 dplyr::collect
使用 dplyr::select
,並依預設列印 title
新 DataFrame 前十個數據列的 、 formatted_date
和資料 day
行:
withUnixTimestamp <- withMMyyyy %>%
mutate(formatted_date = date_format(today, "yyyy-MM-dd"),
day = dayofmonth(formatted_date))
collect(select(withUnixTimestamp, c("title", "formatted_date", "day")))
# A tibble: 100 × 3
# title formatted_date day
# <chr> <chr> <int>
# 1 Things Fall Apart 2022-09-27 27
# 2 Fairy tales 2022-09-27 27
# 3 The Divine Comedy 2022-09-27 27
# 4 The Epic Of Gilgamesh 2022-09-27 27
# 5 The Book Of Job 2022-09-27 27
# 6 One Thousand and One Nights 2022-09-27 27
# 7 Njál's Saga 2022-09-27 27
# 8 Pride and Prejudice 2022-09-27 27
# 9 Le Père Goriot 2022-09-27 27
# 10 Molloy, Malone Dies, The Unnamable, the trilogy 2022-09-27 27
# … with 90 more rows
# ℹ Use `print(n = ...)` to see more rows
您可以在記憶體中建立以現有 DataFrame 為基礎的具名暫存檢視。 例如,在筆記本資料格中執行下列程式碼,以用來 SparkR::createOrReplaceTempView
取得前面名為 jsonTable
DataFrame 的內容,並將暫存檢視從中命名為 timestampTable
。 然後,使用 sparklyr::spark_read_table
來讀取暫存檢視的內容。 使用 sparklyr::collect
預設列印臨時表的前 10 個數據列:
createOrReplaceTempView(withTimestampDF, viewName = "timestampTable")
spark_read_table(
sc = sc,
name = "timestampTable"
) %>% collect()
# A tibble: 100 × 10
# author country image…¹ langu…² link pages title year today
# <chr> <chr> <chr> <chr> <chr> <int> <chr> <int> <dttm>
# 1 Chinua A… Nigeria images… English "htt… 209 Thin… 1958 2022-09-27 21:11:56
# 2 Hans Chr… Denmark images… Danish "htt… 784 Fair… 1836 2022-09-27 21:11:56
# 3 Dante Al… Italy images… Italian "htt… 928 The … 1315 2022-09-27 21:11:56
# 4 Unknown Sumer … images… Akkadi… "htt… 160 The … -1700 2022-09-27 21:11:56
# 5 Unknown Achaem… images… Hebrew "htt… 176 The … -600 2022-09-27 21:11:56
# 6 Unknown India/… images… Arabic "htt… 288 One … 1200 2022-09-27 21:11:56
# 7 Unknown Iceland images… Old No… "htt… 384 Njál… 1350 2022-09-27 21:11:56
# 8 Jane Aus… United… images… English "htt… 226 Prid… 1813 2022-09-27 21:11:56
# 9 Honoré d… France images… French "htt… 443 Le P… 1835 2022-09-27 21:11:56
# 10 Samuel B… Republ… images… French… "htt… 256 Moll… 1952 2022-09-27 21:11:56
# … with 90 more rows, 1 more variable: month <chr>, and abbreviated variable
# names ¹imageLink, ²language
# ℹ Use `print(n = ...)` to see more rows, and `colnames()` to see all variable names
您可以使用 sparklyr 和 dplyr 進行統計分析。
例如,建立要執行統計資料的 DataFrame。 若要這樣做,請在筆記本資料格中執行下列程式碼,以用來 sparklyr::sdf_copy_to
將 R 內建的資料集內容 iris
寫入名為 iris
的資料框架。 使用 sparklyr::sdf_collect
預設列印臨時表的前 10 個數據列:
irisDF <- sdf_copy_to(
sc = sc,
x = iris,
name = "iris",
overwrite = TRUE
)
sdf_collect(irisDF, "row-wise")
# A tibble: 150 × 5
# Sepal_Length Sepal_Width Petal_Length Petal_Width Species
# <dbl> <dbl> <dbl> <dbl> <chr>
# 1 5.1 3.5 1.4 0.2 setosa
# 2 4.9 3 1.4 0.2 setosa
# 3 4.7 3.2 1.3 0.2 setosa
# 4 4.6 3.1 1.5 0.2 setosa
# 5 5 3.6 1.4 0.2 setosa
# 6 5.4 3.9 1.7 0.4 setosa
# 7 4.6 3.4 1.4 0.3 setosa
# 8 5 3.4 1.5 0.2 setosa
# 9 4.4 2.9 1.4 0.2 setosa
# 10 4.9 3.1 1.5 0.1 setosa
# … with 140 more rows
# ℹ Use `print(n = ...)` to see more rows
現在使用 dplyr::group_by
依資料行分組資料 Species
列。 搭配 使用 dplyr::summarize
,依 資料行 Species
的 25th、50th、75 和 100th 分位數 Sepal_Length
計算摘要 dplyr::percentile_approx
統計資料。 使用 sparklyr::collect
列印結果:
備註
dplyr::summarize
只接受符合 Hive 內建函式 (也稱為 UDF) 和內建彙總函式 (也稱為 UDAF) 的引數。 如需一般資訊,請參閱 Hive 函式 。 如需 的相關資訊 percentile_approx
,請參閱 內建彙總函式(UDAF)。
quantileDF <- irisDF %>%
group_by(Species) %>%
summarize(
quantile_25th = percentile_approx(
Sepal_Length,
0.25
),
quantile_50th = percentile_approx(
Sepal_Length,
0.50
),
quantile_75th = percentile_approx(
Sepal_Length,
0.75
),
quantile_100th = percentile_approx(
Sepal_Length,
1.0
)
)
collect(quantileDF)
# A tibble: 3 × 5
# Species quantile_25th quantile_50th quantile_75th quantile_100th
# <chr> <dbl> <dbl> <dbl> <dbl>
# 1 virginica 6.2 6.5 6.9 7.9
# 2 versicolor 5.6 5.9 6.3 7
# 3 setosa 4.8 5 5.2 5.8
例如,您可以使用 sparklyr::sdf_quantile
來計算類似的結果:
print(sdf_quantile(
x = irisDF %>%
filter(Species == "virginica"),
column = "Sepal_Length",
probabilities = c(0.25, 0.5, 0.75, 1.0)
))
# 25% 50% 75% 100%
# 6.2 6.5 6.9 7.9