Trabalhar com DataFrames e tabelas em R
Este artigo descreve como usar pacotes R, como SparkR, sparklyr e dplyr para trabalhar com R data.frames, Spark DataFrames e tabelas na memória.
Observe que, ao trabalhar com SparkR, sparklyr e dplyr, você pode descobrir que pode concluir uma operação específica com todos esses pacotes e pode usar o pacote com o qual se sente mais confortável. Por exemplo, para executar uma consulta, você pode chamar funções como SparkR::sql, sparklyr::sdf_sqle dplyr::select. Em outras ocasiões, você poderá concluir uma operação com apenas um ou dois desses pacotes, e a operação escolhida dependerá do seu cenário de uso. Por exemplo, a maneira como você chama difere ligeiramente da maneira como você chama sparklyr::sdf_quantiledplyr::percentile_approx, embora ambas as funções calcuate quantis.
Você pode usar o SQL como uma ponte entre o SparkR e o sparklyr. Por exemplo, você pode usar SparkR::sql para consultar tabelas que você cria com o sparklyr. Você pode usar sparklyr::sdf_sql para consultar tabelas criadas com o SparkR. E dplyr o código sempre é traduzido para SQL na memória antes de ser executado. Consulte também Interoperabilidade de API e Tradução SQL.
Carregar SparkR, sparklyr e dplyr
Os pacotes SparkR, sparklyr e dplyr estão incluídos no Databricks Runtime instalado nos clusters do Azure Databricks. Portanto, você não precisa chamar o habitual install.package antes de começar a chamar esses pacotes. No entanto, você ainda deve carregar esses pacotes com library primeiro. Por exemplo, de dentro de um bloco de anotações R em um espaço de trabalho do Azure Databricks, execute o seguinte código em uma célula do bloco de anotações para carregar SparkR, sparklyr e dplyr:
library(SparkR)
library(sparklyr)
library(dplyr)
Conectar o sparklyr a um cluster
Depois de carregar o sparklyr, você deve chamar sparklyr::spark_connect para se conectar ao cluster, especificando o método de databricks conexão. Por exemplo, execute o seguinte código em uma célula do bloco de anotações para se conectar ao cluster que hospeda o bloco de anotações:
sc <- spark_connect(method = "databricks")
Por outro lado, um bloco de anotações do Azure Databricks já estabelece um SparkSession no cluster para uso com o SparkR, portanto, você não precisa ligar SparkR::sparkR.session antes de começar a chamar o SparkR.
Carregar um arquivo de dados JSON para seu espaço de trabalho
Muitos dos exemplos de código neste artigo são baseados em dados em um local específico em seu espaço de trabalho do Azure Databricks, com nomes de coluna e tipos de dados específicos. Os dados para este exemplo de código se originam em um arquivo JSON nomeado book.json de dentro do GitHub. Para obter este ficheiro e carregá-lo para a sua área de trabalho:
- Vá para o arquivo books.json no GitHub e use um editor de texto para copiar seu conteúdo para um arquivo nomeado
books.jsonem algum lugar em sua máquina local. - Na barra lateral do espaço de trabalho do Azure Databricks, clique em Catálogo.
- Clique em Criar tabela.
- Na guia Carregar arquivo, solte o
books.jsonarquivo da sua máquina local para a caixa Soltar arquivos para carregar. Ou selecione clique para navegar e navegue até obooks.jsonarquivo de sua máquina local.
Por padrão, o Azure Databricks carrega seu arquivo local para o local books.json DBFS em seu espaço de trabalho com o caminho /FileStore/tables/books.json.
Não clique em Criar tabela com interface do usuário ou Criar tabela no bloco de anotações. Os exemplos de código neste artigo usam os dados no arquivo carregado books.json neste local DBFS.
Leia os dados JSON em um DataFrame
Use sparklyr::spark_read_json para ler o arquivo JSON carregado em um DataFrame, especificando a conexão, o caminho para o arquivo JSON e um nome para a representação de tabela interna dos dados. Para este exemplo, você deve especificar que o book.json arquivo contém várias linhas. Especificar o esquema das colunas aqui é opcional. Caso contrário, o sparklyr infere o esquema das colunas por padrão. Por exemplo, execute o seguinte código em uma célula do bloco de anotações para ler os dados do arquivo JSON carregado em um DataFrame chamado 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"
)
)
Imprimir as primeiras linhas de um DataFrame
Você pode usar SparkR::head, SparkR::showou sparklyr::collect para imprimir as primeiras linhas de um DataFrame. Por padrão, head imprime as seis primeiras linhas por padrão. show e imprima collect as primeiras 10 linhas. Por exemplo, execute o seguinte código em uma célula do bloco de anotações para imprimir as primeiras linhas do DataFrame chamado 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
Executar consultas SQL, gravar e ler a partir de uma tabela
Você pode usar funções dplyr para executar consultas SQL em um DataFrame. Por exemplo, execute o código a seguir em uma célula do bloco de anotações para usar dplyr::group_by e dployr::count obter contagens por autor do DataFrame chamado jsonDF. Use dplyr::arrange e dplyr::desc classifique o resultado em ordem decrescente por contagens. Em seguida, imprima as primeiras 10 linhas por padrão.
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
Em seguida, você pode usar sparklyr::spark_write_table para gravar o resultado em uma tabela no Azure Databricks. Por exemplo, execute o seguinte código em uma célula do bloco de anotações para executar novamente a consulta e, em seguida, escreva o resultado em uma tabela chamada json_books_agg:
group_by(jsonDF, author) %>%
count() %>%
arrange(desc(n)) %>%
spark_write_table(
name = "json_books_agg",
mode = "overwrite"
)
Para verificar se a tabela foi criada, você pode usar sparklyr::sdf_sql junto com SparkR::showDF para exibir os dados da tabela. Por exemplo, execute o seguinte código em uma célula do bloco de anotações para consultar a tabela em um DataFrame e, em seguida, use sparklyr::collect para imprimir as primeiras 10 linhas do DataFrame por padrão:
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
Você também pode usar sparklyr::spark_read_table para fazer algo semelhante. Por exemplo, execute o seguinte código em uma célula do bloco de anotações para consultar o DataFrame anterior nomeado jsonDF em um DataFrame e, em seguida, use sparklyr::collect para imprimir as primeiras 10 linhas do DataFrame por padrão:
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
Adicionar colunas e calcular valores de coluna em um DataFrame
Você pode usar funções dplyr para adicionar colunas a DataFrames e calcular valores de colunas.
Por exemplo, execute o código a seguir em uma célula do bloco de anotações para obter o conteúdo do DataFrame chamado jsonDF. Use dplyr::mutate para adicionar uma coluna chamada today, e preencha esta nova coluna com o carimbo de data/hora atual. Em seguida, escreva esse conteúdo em um novo DataFrame chamado withDate e use dplyr::collect para imprimir as primeiras 10 linhas do novo DataFrame por padrão.
Nota
dplyr::mutate só aceita argumentos que estejam em conformidade com as funções internas do Hive (também conhecidas como UDFs) e funções agregadas internas (também conhecidas como UDAFs). Para obter informações gerais, consulte Funções do Hive. Para obter informações sobre as funções relacionadas à data nesta seção, consulte Funções de data.
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
Agora use dplyr::mutate para adicionar mais duas colunas ao conteúdo do withDate DataFrame. As colunas new month e contêm o mês e year o ano numéricos da today coluna. Em seguida, escreva esse conteúdo em um novo DataFrame chamado withMMyyyy, e use dplyr::select junto com dplyr::collect para imprimir as authorcolunas , titlemonth e year das primeiras dez linhas do novo DataFrame por padrão:
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
Agora use dplyr::mutate para adicionar mais duas colunas ao conteúdo do withMMyyyy DataFrame. As novas formatted_date colunas contêm a parte da today coluna, enquanto a yyyy-MM-dd nova coluna contém o dia numérico da nova dayformatted_date coluna. Em seguida, escreva esse conteúdo em um novo DataFrame chamado withUnixTimestamp, e use dplyr::select junto com dplyr::collect para imprimir as titlecolunas , formatted_datee day das primeiras dez linhas do novo DataFrame por padrão:
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
Criar uma vista temporária
Você pode criar exibições temporárias nomeadas na memória baseadas em DataFrames existentes. Por exemplo, execute o código a seguir em uma célula do bloco de anotações para usar SparkR::createOrReplaceTempView para obter o conteúdo do DataFrame anterior nomeado jsonTable e fazer uma exibição temporária dele chamado timestampTable. Em seguida, use sparklyr::spark_read_table para ler o conteúdo do modo de exibição temporário. Use sparklyr::collect para imprimir as primeiras 10 linhas da tabela temporária por padrão:
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
Executar análise estatística em um DataFrame
Você pode usar o sparklyr junto com o dplyr para análises estatísticas.
Por exemplo, crie um DataFrame para executar estatísticas. Para fazer isso, execute o seguinte código em uma célula do bloco de anotações para usar sparklyr::sdf_copy_to para gravar o conteúdo do conjunto de iris dados que é construído em R para um DataFrame chamado iris. Use sparklyr::sdf_collect para imprimir as primeiras 10 linhas da tabela temporária por padrão:
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
Agora use dplyr::group_by para agrupar linhas pela Species coluna. Use dplyr::summarize junto com dplyr::percentile_approx para calcular estatísticas resumidas pelos quantis 25, 50, 75 e 100 da Sepal_Length coluna por Species. Imprima sparklyr::collect os resultados:
Nota
dplyr::summarize só aceita argumentos que estejam em conformidade com as funções internas do Hive (também conhecidas como UDFs) e funções agregadas internas (também conhecidas como UDAFs). Para obter informações gerais, consulte Funções do Hive. Para obter informações sobre percentile_approxo , consulte UDAF (Built-in Aggregate Functions).
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
Resultados semelhantes podem ser calculados, por exemplo, usando 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