R에서 DataFrame 및 테이블 작업
이 문서에서는 SparkR, sparklyr, dplyr와 같은 R 패키지를 사용하여 R data.frames, Spark DataFrames, 메모리 내 테이블을 사용하는 방법을 설명합니다.
SparkR, sparklyr, dplyr로 작업할 때 이러한 모든 패키지로 특정 작업을 완료할 수 있으며 가장 익숙한 패키지를 사용할 수 있습니다. 예를 들어 쿼리를 실행하려면 SparkR::sql, sparklyr::sdf_sql, dplyr::select와 같은 함수를 호출할 수 있습니다. 다른 경우에는 이러한 패키지 중 하나 또는 두 개만 사용하여 작업을 완료할 수 있으며 선택한 작업은 사용 시나리오에 따라 달라집니다. 예를 들어 sparklyr::sdf_quantile을 호출하는 방법은 두 함수가 양자를 계산하더라도 dplyr::percentile_approx를 호출하는 방식과 약간 다릅니다.
SPARKR과 sparklyr 사이의 브리지로 SQL을 사용할 수 있습니다. 예를 들어 SparkR::sql을 사용하여 sparklyr로 만든 테이블을 쿼리할 수 있습니다. sparklyr::sdf_sql을 사용하여 SparkR로 만든 테이블을 쿼리할 수 있습니다. 그리고 dplyr 코드는 실행되기 전에 항상 메모리에서 SQL로 변환됩니다. API 상호 운용성 및 SQL 번역도 참조하세요.
SparkR, sparklyr, dplyr 로드
SparkR, sparklyr, dplyr 패키지는 Azure Databricks 클러스터에 설치된 Databricks Runtime에 포함됩니다. 따라서 이러한 패키지 호출을 시작하기 전에 일반적인 install.package를 호출할 필요가 없습니다. 그러나 먼저 library를 사용하여 이러한 패키지를 로드해야 합니다. 예를 들어 Azure Databricks 작업 영역의 R Notebook 내의 Notebook 셀에서 다음 코드를 실행하여 SparkR, sparklyr, dplyr를 로드합니다.
library(SparkR)
library(sparklyr)
library(dplyr)
클러스터에 sparklyr 연결
sparklyr를 로드한 후 sparklyr::spark_connect를 호출하여 databricks 연결 방법을 지정하고 클러스터에 연결해야 합니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하여 Notebook을 호스트하는 클러스터에 연결합니다.
sc <- spark_connect(method = "databricks")
반면, Azure Databricks Notebook은 SparkR에서 사용하기 위해 클러스터에 이미 SparkSession를 설정하므로 SparkR 호출을 시작하기 전에 SparkR::sparkR.session를 호출할 필요가 없습니다.
작업 영역에 JSON 데이터 파일 업로드
이 문서의 많은 코드 예제는 특정 열 이름 및 데이터 형식을 사용하여 Azure Databricks 작업 영역의 특정 위치에 있는 데이터를 기반으로 합니다. 이 코드 예제의 데이터는 GitHub 내에서 book.json이라는 JSON 파일에서 시작됩니다. 이 파일을 가져와 작업 영역에 업로드하려면 다음을 수행합니다.
- GitHub의 books.json 파일로 이동하고 텍스트 편집기를 사용하여 해당 콘텐츠를 로컬 컴퓨터의 어딘가에 있는
books.json파일에 복사합니다. - Azure Databricks 작업 영역 사이드바에서 카탈로그를 클릭합니다.
- 테이블 만들기를 클릭합니다.
- 파일 업로드 탭에서 로컬 컴퓨터의
books.json파일을 업로드할 파일 놓기 상자에 놓습니다. 또는 클릭하여 찾아보기를 선택하고 로컬 컴퓨터에서books.json파일을 찾습니다.
기본적으로 Azure Databricks는 /FileStore/tables/books.json 경로를 사용하여 작업 영역의 DBFS 위치에 로컬 books.json 파일을 업로드합니다.
UI를 사용하여 테이블 만들기 또는 Notebook에 테이블 만들기는 클릭하지 마세요. 이 문서의 코드 예제에서는 이 DBFS 위치에 업로드된 books.json 파일의 데이터를 사용합니다.
DataFrame으로 JSON 데이터 읽기
sparklyr::spark_read_json를 사용하여 업로드된 JSON 파일을 DataFrame으로 읽고 연결, JSON 파일의 경로, 데이터의 내부 테이블 표현에 대한 이름을 지정합니다. 이 예제에서는 book.json 파일에 여러 줄이 포함되도록 지정해야 합니다. 여기에서 열의 스키마를 지정하는 것은 선택 사항입니다. 그렇지 않으면 sparklyr는 기본적으로 열의 스키마를 유추합니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하여 업로드된 JSON 파일의 데이터를 jsonDF라는 DataFrame으로 읽습니다.
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"
)
)
DataFrame의 처음 몇 행 인쇄
SparkR::head, SparkR::show 또는 sparklyr::collect를 사용하여 DataFrame의 첫 번째 행을 인쇄할 수 있습니다. 기본적으로 head는 처음 6개의 행을 기본적으로 인쇄합니다. show 및 collect는 처음 10개 행을 인쇄합니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하여 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
SQL 쿼리를 실행하고, 테이블에 쓰고, 테이블에서 읽습니다.
dplyr 함수를 사용하여 DataFrame에서 SQL 쿼리를 실행할 수 있습니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하여 및 dplyr::group_bydployr::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의 테이블에 결과를 쓸 수 있습니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하여 쿼리를 다시 실행한 다음, 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을 사용하여 테이블의 데이터를 표시할 수 있습니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하여 테이블을 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을 사용하여 비슷한 작업을 수행할 수도 있습니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하여 jsonDF라는 앞의 DataFrame을 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
DataFrame에서 열 및 컴퓨팅 열 값 추가
dplyr 함수를 사용하여 DataFrame에 열을 추가하고 열 값을 계산할 수 있습니다.
예를 들어 Notebook 셀에서 다음 코드를 실행하여 jsonDF라는 DataFrame의 내용을 가져옵니다. dplyr::mutate를 사용하여 today라는 열을 추가하고 이 새 열을 현재 타임스탬프로 채웁니다. 그런 다음, 이러한 내용을 withDate라는 새 DataFrame에 쓰고 dplyr::collect를 사용하여 기본적으로 새 DataFrame의 처음 10개 행을 인쇄합니다.
참고 항목
dplyr::mutate는 Hive의 기본 제공 함수(UDF라고도 함) 및 기본 제공 집계 함수(UDAF라고도 함)를 준수하는 인수만 허용합니다. 일반적인 내용은 Hive 함수를 참조하세요. 이 섹션의 날짜 관련 함수에 대한 자세한 내용은 날짜 함수를 참조하세요.
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를 사용하여 withDate DataFrame의 내용에 두 개의 열을 더 추가합니다. 새 month 열과 year 열에는 today 열의 숫자 월과 연도가 포함됩니다. 그런 다음, 이러한 내용을 withMMyyyy라는 새 DataFrame에 쓰고 dplyr::collect와 함께 dplyr::select를 사용하여 기본적으로 새 DataFrame의 처음 10개 행의 author, 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를 사용하여 withMMyyyy DataFrame의 내용에 두 개의 열을 더 추가합니다. 새 formatted_date 열에는 today 열의 yyyy-MM-dd 부분이 포함되며 새 day 열에는 새 formatted_date 열의 숫자 날짜가 포함됩니다. 그런 다음, 이러한 내용을 withUnixTimestamp라는 새 DataFrame에 쓰고 dplyr::collect와 함께 dplyr::select를 사용하여 기본적으로 새 DataFrame의 처음 10개 행의 title, 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을 기반으로 하는 메모리에 명명된 임시 보기를 만들 수 있습니다. 예를 들어 Notebook 셀에서 다음 코드를 실행하고 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
DataFrame에 대한 통계 분석 수행
통계 분석에 dplyr와 함께 sparklyr를 사용할 수 있습니다.
예를 들어 통계를 실행할 DataFrame을 만듭니다. 이렇게 하려면 Notebook 셀에서 다음 코드를 실행하고 sparklyr::sdf_copy_to를 사용하여 R에 기본 제공되는 iris 데이터 세트의 내용을 iris라는 DataFrame에 씁니다. 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::percentile_approx와 함께 dplyr::summarize를 사용하여 Species의 Sepal_Length 열 25번째, 50번째, 75번째, 100번째 분위수까지 요약 통계를 계산합니다. 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