SparkR로 큰 데이터를 줄이기

Code Workbook은 SparkR의 분산 SparkDataFrame과 로컬 R data.frame을 구분합니다. 큰 입력은 SparkR로 열 선택, 필터와 집계를 수행한 뒤 드라이버 메모리에 들어갈 결과만 collect()하는 것이 권장됩니다.

아래 예제는 R 콘솔의 Variables에서 df라는 입력 노드를 Spark dataframe으로 설정한 상황입니다. 각 예제에서 참조한 열이 해당 입력에 있어야 합니다. numeric_col, col1, col2는 수치 열이며 그룹용 열은 group_col1, group_col2입니다.

행 필터

SQL의 WHERE와 같은 문자열 조건을 사용할 수 있습니다.

filtered_sql <- SparkR::filter(df, "numeric_col > 10")

같은 조건을 Column 표현식으로 쓸 수도 있습니다.

filtered_column <- SparkR::filter(df, df$numeric_col > 10)

where()도 같은 필터 목적에 사용할 수 있습니다.

filtered_where <- SparkR::where(df, "numeric_col > 10")

열 선택과 이름 변경

필요한 열만 선택합니다.

subset_columns <- SparkR::select(df, "column1", "column2", "column3")

열 이름을 바꿀 때는 새 DataFrame을 받습니다.

renamed <- SparkR::withColumnRenamed(
    df, "old_column_name", "new_column_name"
)

열 계산

두 수치 열의 합이나 상수를 곱한 결과를 추가할 수 있습니다.

with_sum <- SparkR::withColumn(df, "col1_plus_col2", df$col1 + df$col2)
with_scaled <- SparkR::withColumn(df, "col1_times_60", df$col1 * 60)

원천 DataFrame의 열을 인자로 사용하는 연산이며 각 변수는 새로운 계산 결과를 가리킵니다.

그룹 집계

groupBy()는 그룹 정의를 만들고 agg()가 실제 집계 DataFrame을 구성합니다. 평균과 최댓값을 같은 그룹에 계산할 수 있습니다.

grouped <- SparkR::groupBy(df, "group_col1", "group_col2")
aggregated <- SparkR::agg(
    grouped,
    average_col1 = SparkR::avg(df$col1),
    max_col = SparkR::max(df$col1)
)

결과에는 그룹 열과 average_col1, max_col이 들어갑니다. 충분히 작게 줄인 결과를 로컬 R 작업으로 이어가려면 다음과 같이 수집합니다.

local_result <- SparkR::collect(aggregated)

collect()는 모든 결과 행을 드라이버로 가져옵니다. 집계를 했다는 이유만으로 결과가 항상 작은 것은 아니므로 그룹 수와 메모리 크기를 함께 고려합니다. 언어 활성화와 R 유형 변환 제한은 언어 및 입출력의 조건을 따릅니다.