Java Spark SQL 연산 참조

Java의 표 데이터는 Dataset<Row>입니다. 아래 식의 df, lookup, other는 계산 함수가 받은 DataFrame이며, 참조한 열이 있어야 합니다. 일반적인 import는 다음과 같습니다.

import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.DataTypes;
import static org.apache.spark.sql.functions.*;

필터와 정렬

isin은 비교값들을 varargs로 받습니다. Java List를 값 하나로 전달하지 말고 여러 인자 또는 해당 값들의 Object 배열을 전달합니다. 조건에 사용하는 열과 값의 자료형도 맞아야 합니다.

조인

양쪽에 같은 이름의 키가 있으면 using-column 오버로드를 사용합니다.

Dataset<Row> leftJoined = df.join(lookup, "person_id", "left");
Dataset<Row> unmatched = df.join(lookup, "person_id", "leftanti");

leftanti는 오른쪽에 일치하는 행이 없는 왼쪽 행만 반환합니다. col("person_id") 자체를 Boolean 조인 조건으로 사용하지 않습니다.

열 이름이 다르면 각 DataFrame의 열로 조건을 만듭니다.

Dataset<Row> joined = df.join(
        other, df.col("id").equalTo(other.col("person_id")), "left");
Dataset<Row> byName = df.join(
        other,
        df.col("first_name").equalTo(other.col("name"))
                .and(df.col("last_name").equalTo(other.col("family_name"))),
        "left");

양쪽에 중복 키가 있으면 일치하는 조합만큼 결과 행이 늘어날 수 있습니다.

조회값으로 키 대체

다음은 조회값이 있으면 왼쪽 키 열을 그 값으로 바꾸고, 없으면 원래 키를 남기는 helper입니다. 키와 조회값은 coalesce가 함께 반환할 수 있는 호환 자료형이어야 합니다. 오른쪽 키가 중복되면 결과도 여러 행으로 늘어나며 자동 중복 제거를 하지 않습니다.

파일 myproject/LookupColumns.java:

package myproject;

import static org.apache.spark.sql.functions.coalesce;

import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class LookupColumns {
    private LookupColumns() {}

    public static Dataset<Row> lookupAndReplace(
            Dataset<Row> source,
            Dataset<Row> lookup,
            String sourceKey,
            String lookupKey,
            String lookupValue) {
        String[] names = source.columns();
        Dataset<Row> left = source.alias("source_rows");
        Dataset<Row> right = lookup.select(
                lookup.col(lookupKey).alias("lookup_key"),
                lookup.col(lookupValue).alias("lookup_value"))
                .alias("lookup_rows");
        Dataset<Row> joined = left.join(
                right, left.col(sourceKey).equalTo(right.col("lookup_key")), "left");
        Column[] projection = new Column[names.length];
        for (int index = 0; index < names.length; index++) {
            String name = names[index];
            projection[index] = name.equals(sourceKey)
                    ? coalesce(right.col("lookup_value"), left.col(name)).alias(name)
                    : left.col(name);
        }
        return joined.select(projection);
    }
}

계산 함수에서 people과 payCodes를 받은 경우 호출 예는 LookupColumns.lookupAndReplace(people, payCodes, "pay_code", "code", "description")입니다. 이 예에서 pay_code, code와 description은 문자열 열로 구성합니다.

열 추가, 선택과 이름

두 데이터셋에 공통인 열

다음 helper들은 현재 Java API의 Column 배열을 사용하므로 Scala collection 변환이 필요하지 않습니다.

package myproject;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Set;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class ColumnTools {
    private ColumnTools() {}

    public static Dataset<Row> commonColumns(Dataset<Row> source, Dataset<Row> other) {
        Set<String> names = new HashSet<>();
        Collections.addAll(names, other.columns());
        List<Column> selected = new ArrayList<>();
        for (String name : source.columns()) {
            if (names.contains(name)) {
                selected.add(source.col(name));
            }
        }
        return source.select(selected.toArray(new Column[selected.size()]));
    }

    public static Dataset<Row> normalizeNames(Dataset<Row> source) {
        String[] names = source.columns();
        for (int index = 0; index < names.length; index++) {
            names[index] = names[index].toLowerCase(Locale.ROOT)
                    .replace(' ', '_').replace('-', '_');
        }
        return source.toDF(names);
    }
}

이름 정규화는 소문자화하고 공백 및 하이픈을 밑줄로 바꿉니다. 서로 다른 원래 이름이 같은 결과가 되지 않는지 확인해야 합니다. 공통 열이 하나도 없으면 표 데이터셋 출력에 필요한 스키마를 별도로 정해야 합니다.

자료형, null과 중복

문자열 필터

이 조건들을 df.filter(...)에 전달합니다.

문자열 계산

결과 열을 저장하려면 withColumn("열 이름", 식)에 사용합니다.

수치 연산

날짜와 시각

다음 가상 날짜 범위는 양끝을 포함합니다.

Dataset<Row> inRange = df.filter(
        col("date_of_birth").geq("2017-05-10")
                .and(col("date_of_birth").leq("2018-07-21")));

잘못된 문자열의 파싱이나 범위 밖 값은 Spark 버전 및 ANSI 설정에 따라 오류나 null 처리 차이가 있으므로 정상 입력 형식과 실행 설정을 맞춥니다.

Array, Map과 Struct

explode는 array나 map에 사용합니다. Struct 자체를 여러 행으로 explode하는 연산이 아니며 struct 필드를 열로 펼칠 때는 struct_name.*를 사용합니다. Map에서 없는 키를 조회하면 null이 될 수 있지만 struct의 없는 필드와는 다른 계약입니다.

집계

Dataset<Row> residents = df.groupBy(col("address")).agg(
        count(col("uuid")).alias("num_residents"),
        max(col("age")).alias("oldest_age"),
        first(col("city"), true).alias("city"));
Dataset<Row> names = df.groupBy(col("address")).agg(
        collect_set(col("name")).alias("resident_names"));

num_residents를 실제 주민 행 수로 해석하려면 uuid가 null이 아닌 입력이어야 합니다.

파티션과 로그

df.repartition(1)은 파티션 하나로 재분배합니다. 큰 데이터의 쓰기 병목과 메모리 사용은 출력 파일 준비를 고려합니다.

SLF4J 로그는 클래스의 필드에 logger를 두고 실제 메서드 안에서 호출합니다. 아래 전체 변환 예제도 이 방식을 사용합니다.

Decimal을 받는 Java Spark UDF

이는 Builder에 게시하는 Foundry UDF와 달리 Spark 표현식의 Java UDF입니다. 아래 입력 a, b는 Decimal 열이며 Java에서는 BigDecimal로 받습니다. Null 입력을 명시적으로 처리합니다.

package myproject.datasets;

import static org.apache.spark.sql.functions.col;
import static org.apache.spark.sql.functions.udf;

import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;
import java.math.BigDecimal;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.api.java.UDF2;
import org.apache.spark.sql.expressions.UserDefinedFunction;
import org.apache.spark.sql.types.DecimalType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class DecimalAddition {
    private static final Logger LOG = LoggerFactory.getLogger(DecimalAddition.class);

    @Compute
    @Output("/Example Project/Outputs/decimal_sum")
    public Dataset<Row> compute(
            @Input("/Example Project/Inputs/decimals") Dataset<Row> input) {
        LOG.info("Creating decimal addition expression");
        UDF2<BigDecimal, BigDecimal, BigDecimal> add = (left, right) ->
                left == null || right == null ? null : left.add(right);
        UserDefinedFunction expression = udf(add, new DecimalType());
        return input.withColumn("a_plus_b", expression.apply(col("a"), col("b")));
    }
}

인자 없는 DecimalType()의 기본값은 precision 10, scale 0입니다. 소수 자릿수나 더 큰 범위가 필요하면 입력과 기대 결과에 맞는 precision 및 scale을 명시해야 합니다. UDF2를 명시하여 Java 함수와 다른 언어 함수 오버로드 사이의 선택도 분명하게 합니다.