Práce s datovými rámci (DataFrames) a tables v jazyce R
Důležitý
SparkR v Databricks je zastaralý ve verzi Databricks Runtime 16.0 a novější. Databricks místo toho doporučuje používat sparklyr.
Tento článek popisuje, jak používat balíčky R, jako jsou SparkR, sparklyra dplyr pro práci s R data.frames, Spark DataFramesa tablesv paměti .
Mějte na paměti, že při práci se SparkR, sparklyr a dplyr můžete zjistit, že můžete dokončit konkrétní operaci se všemi těmito balíčky a můžete použít balíček, se kterým jste nejpohodlnější. Například ke spuštění dotazu můžete volat funkce, jako jsou SparkR::sql, sparklyr::sdf_sqla dplyr::select. Jindy může být možné dokončit operaci jenom s jedním nebo dvěma z těchto balíčků a operace, kterou zvolíte, závisí na vašem scénáři použití. Například způsob volání sparklyr::sdf_quantile se mírně liší od způsobu volání dplyr::percentile_approx, i když obě funkce počítají kvantily.
SQL můžete použít jako most mezi SparkR a sparklyr. Pomocí SparkR::sql můžete například dotazovat tables, které jste vytvořili pomocí sparklyr. K dotazování tables, které vytvoříte pomocí SparkR, můžete použít sparklyr::sdf_sql. A dplyr kód se před spuštěním vždy přeloží do SQL v paměti. Viz také interoperabilita rozhraní API a překlad SQL.
Načtěte SparkR, sparklyr a dplyr
Balíčky SparkR, sparklyr a dplyr jsou součástí modulu Databricks Runtime, který je nainstalovaný v clusterech Azure Databricks . Proto není nutné volat obvyklé install.package před zahájením volání těchto balíčků. Tyto balíčky však musíte nejprve načíst s library. Například z poznámkového bloku R v pracovním prostoru Azure Databricks spusťte v buňce poznámkového bloku následující kód, který načte SparkR, sparklyr a dplyr:
library(SparkR)
library(sparklyr)
library(dplyr)
Připojení sparklyru ke clusteru
Po načtení sparklyru je třeba nejprve zavolat sparklyr::spark_connect, abyste se připojili ke clusteru, a určit metodu připojení databricks. Například spuštěním následujícího kódu v buňce poznámkového bloku se připojte ke clusteru, který je hostitelem poznámkového bloku:
sc <- spark_connect(method = "databricks")
Oproti tomu Azure Databricks již vytváří SparkSession v clusteru pro použití se SparkR, takže před zahájením volání SparkR nemusíte volat SparkR::sparkR.session.
Nahrání datového souboru JSON do pracovního prostoru
Mnoho příkladů kódu v tomto článku vychází z dat v určitém umístění v pracovním prostoru Azure Databricks s konkrétními názvy column a datovými typy. Data pro tento příklad kódu pocházejí ze souboru JSON s názvem book.json z GitHubu. Pokud chcete tento soubor get a nahrát ho do pracovního prostoru:
- Přejděte do souboru books.json na GitHubu a pomocí textového editoru zkopírujte jeho obsah do souboru s názvem
books.jsonněkde na místním počítači. - Na bočním panelu pracovního prostoru Azure Databricks klikněte na Catalog.
- Klikněte na Vytvořit Table.
- Na kartě Nahrát soubor přetáhněte soubor
books.jsonz místního počítače do pole Drop files a nahrajte pole. Nebo selectkliknutím procházeta přejít na souborbooks.jsonz místního počítače.
Ve výchozím nastavení Azure Databricks nahraje místní soubor books.json do umístění DBFS ve vašem pracovním prostoru s cestou /FileStore/tables/books.json.
Neklikejte na Vytvořit Table pomocí uživatelského rozhraní nebo Vytvořit Table v poznámkovém bloku. Příklady kódu v tomto článku používají data z nahraného souboru books.json uloženého v tomto umístění DBFS.
Čtení dat JSON do datového rámce
Pomocí sparklyr::spark_read_json načtěte nahraný soubor JSON do datového rámce, zadejte připojení, cestu k souboru JSON a název pro interní reprezentaci dat table. V tomto příkladu musíte určit, že soubor book.json obsahuje více řádků. Zadání columnsschema je volitelné. Jinak sparklyr ve výchozím nastavení odvodí columns' schema. Například spuštěním následujícího kódu v buňce poznámkového bloku načtěte data nahraného souboru JSON do datového rámce s názvem 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"
)
)
Tisk prvních několika řádků datového rámce
První řádky datového rámce můžete vytisknout pomocí SparkR::head, SparkR::shownebo sparklyr::collect. Ve výchozím nastavení head vytiskne prvních šest řádků.
show a collect vytisknou prvních 10 řádků. Například spuštěním následujícího kódu v buňce poznámkového bloku vytiskněte první řádky datového rámce s názvem jsonDF:
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
Spouštění dotazů SQL a zápis do a čtení z table
Pomocí funkcí dplyr můžete spouštět dotazy SQL na datovém rámci. Pro spuštění následujícího kódu v buňce poznámkového bloku použijte dplyr::group_by a dployr::count k zobrazení počtů get podle autora z datového rámce s názvem jsonDF. Pomocí dplyr::arrange a dplyr::desc seřaďte výsledek sestupně podle počtu. Potom ve výchozím nastavení vytiskněte prvních 10 řádků.
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
Poté můžete použít sparklyr::spark_write_table k zápisu do table v Azure Databricks. Spuštěním následujícího kódu v buňce poznámkového bloku spusťte dotaz znovu a potom napište výsledek do table s názvem json_books_agg:
group_by(jsonDF, author) %>%
count() %>%
arrange(desc(n)) %>%
spark_write_table(
name = "json_books_agg",
mode = "overwrite"
)
K ověření vytvoření table můžete použít sparklyr::sdf_sql společně s SparkR::showDF k zobrazení dat table. Například spuštěním následujícího kódu v buňce poznámkového bloku zadejte dotaz na table do datového rámce a pak pomocí sparklyr::collect vytiskněte prvních 10 řádků datového rámce ve výchozím nastavení:
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
Můžete také použít sparklyr::spark_read_table, abyste udělali něco podobného. Například spuštěním následujícího kódu v buňce poznámkového bloku zadejte dotaz na předchozí datový rámec s názvem jsonDF do datového rámce a potom pomocí sparklyr::collect vytiskněte prvních 10 řádků datového rámce ve výchozím nastavení:
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
Přidejte columns a vypočítejte columnvalues v datovém rámci
Funkce dplyr můžete použít k přidání columns do datových rámců a k výpočtu columns' values.
Například spuštěním následujícího kódu v buňce poznámkového bloku get obsah datového rámce s názvem jsonDF. Pomocí dplyr::mutate přidejte column pojmenovanou todaya vyplňte tuto novou column aktuálním časovým razítkem. Potom tento obsah zapište do nového datového rámce s názvem withDate a pomocí dplyr::collect ve výchozím nastavení vytiskněte prvních 10 řádků nového datového rámce.
Poznámka
dplyr::mutate přijímá pouze argumenty, které odpovídají předdefinovaným funkcím Hive (označovaným také jako UDF) a integrovaným agregačním funkcím (označovaným také jako UDAF). Pro informace o funkci Hive, podívejte se na . Informace o funkcích souvisejících s datem v této části najdete v tématu Funkce kalendářních dat.
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
Teď pomocí dplyr::mutate přidejte do obsahu datového rámce withDate další columns. Nové month a yearcolumns obsahují číselný měsíc a rok z todaycolumn. Potom tento obsah napište do nového datového rámce s názvem withMMyyyya pomocí dplyr::select společně s dplyr::collect vytiskněte ve výchozím nastavení author, title, month a yearcolumns prvních deset řádků nového datového rámce:
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
Teď pomocí dplyr::mutate přidejte do obsahu datového rámce withMMyyyy další columns. Nový formatted_datecolumns obsahuje část yyyy-MM-dd z todaycolumn, zatímco nový daycolumn obsahuje číselný den z nového formatted_datecolumn. Potom napište tento obsah do nového datového rámce s názvem withUnixTimestampa použijte dplyr::select spolu s dplyr::collect k tisku title, formatted_datea daycolumns prvních deset řádků nového datového rámce ve výchozím nastavení:
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
Vytvoření dočasného zobrazení
V paměti můžete vytvořit pojmenované dočasné views, které jsou založené na existujících datových rámcích. Například spuštěním následujícího kódu v buňce poznámkového bloku použijte SparkR::createOrReplaceTempView k get obsah předchozího datového rámce s názvem jsonTable a z něj vytvořte dočasné zobrazení s názvem timestampTable. Potom použijte sparklyr::spark_read_table ke čtení obsahu dočasného zobrazení. Pomocí sparklyr::collect můžete ve výchozím nastavení vytisknout prvních 10 řádků dočasného table:
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
Provádění statistické analýzy datového rámce
Sparklyr můžete použít společně s dplyr pro statistické analýzy.
Můžete například vytvořit datový rámec pro spouštění statistik. Uděláte to tak, že v buňce poznámkového bloku spustíte následující kód, který použije sparklyr::sdf_copy_to k zápisu obsahu datové sady iris, která je integrovaná do datového rámce R s názvem iris. Pomocí sparklyr::sdf_collect můžete ve výchozím nastavení vytisknout prvních 10 řádků dočasného table:
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
Teď pomocí dplyr::group_by seskupit řádky podle Speciescolumn. S dplyr::summarize spolu s dplyr::percentile_approx můžete vypočítat souhrnné statistiky pro 25., 50., 75. a 100. kvantilů Sepal_Lengthcolumn podle Species. Použijte sparklyr::collect a vytiskněte výsledky.
Poznámka
dplyr::summarize přijímá pouze argumenty, které odpovídají předdefinovaným funkcím Hive (označovaným také jako UDF) a integrovaným agregačním funkcím (označovaným také jako UDAF). Obecné informace najdete v části funkce Hive. Informace o percentile_approxnaleznete v tématu předdefinované agregační funkce (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
Podobné výsledky se dají vypočítat například pomocí 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