Workbook에서 비정형 파일 읽고 쓰기

데이터셋의 표 스키마 대신 실제 파일을 읽으면 이미지, XML, JSON, GZIP, ZIP, RDA와 RDS 등을 처리할 수 있습니다. 파일 입력은 가져온 데이터셋 또는 저장된 파생 데이터셋이어야 합니다. 저장하지 않은 변환의 메모리 결과에는 파일 입력 API를 사용할 수 없습니다.

Python 파일 입력

Inputs에서 Python transform input을 선택하면 filesystem()으로 데이터셋 내부의 파일 경로에 접근할 수 있습니다. 스키마 없는 데이터셋은 기본적으로 이 유형으로 읽습니다. 입력 핸들은 브랜치와 RID 같은 메타데이터도 제공합니다.

ZIP 안의 CSV를 표로 읽기

다음 예제의 입력 alias는 datasetOfZippedFiles이며 유형은 Python transform input입니다. 데이터셋의 ZIP들 안에 UTF-8 CSV가 있고 각 CSV의 헤더는 id,name이라고 가정합니다. 반환값은 두 String 열의 Spark DataFrame입니다.

def sample(datasetOfZippedFiles):
    import csv
    import io
    import shutil
    import tempfile
    import zipfile

    fs = datasetOfZippedFiles.filesystem()

    def process_file(file_status):
        with fs.open(file_status.path, "rb") as source:
            with tempfile.NamedTemporaryFile() as local_zip:
                shutil.copyfileobj(source, local_zip)
                local_zip.flush()
                local_zip.seek(0)
                with zipfile.ZipFile(local_zip) as archive:
                    for member in archive.infolist():
                        if member.is_dir() or not member.filename.lower().endswith(".csv"):
                            continue
                        with archive.open(member) as binary:
                            with io.TextIOWrapper(
                                binary, encoding="utf-8", newline=""
                            ) as text:
                                for row in csv.DictReader(text):
                                    yield (row["id"], row["name"])

    rows = fs.files().rdd.flatMap(process_file)
    return spark.createDataFrame(rows, schema="id string, name string")

Workbook이 제공한 spark session을 사용합니다. ZIP 읽기는 seek 가능한 파일이 필요하므로 각 압축 파일을 임시 디스크에 복사합니다. 압축 안의 CSV 내용은 행 단위로 읽고, CSV parser가 따옴표 안 쉼표와 줄바꿈을 처리합니다. 입력 파일별 작업은 Spark에 분산할 수 있지만 하나의 ZIP 내부 순회를 여러 task로 자동 분할하는 구성은 아닙니다.

Python 출력 핸들

파일을 쓰는 노드는 반드시 Save as dataset을 켜야 합니다. 콘솔에서는 TransformOutput으로 파일을 쓸 수 없습니다.

Workbook 런타임의 Transforms.get_output()으로 출력 핸들을 얻습니다. 생성한 핸들의 filesystem() 또는 쓰기 메서드를 사용한 뒤 다른 일반 값을 반환해도 그 값은 별도 출력으로 사용되지 않습니다. 아래처럼 사용한 핸들을 반환하여 출력 경로를 명확히 합니다.

텍스트 파일

def write_text_file():
    output = Transforms.get_output()
    with output.filesystem().open("my text file.txt", "w") as handle:
        handle.write("Hello world")
    return output

분할된 CSV 데이터셋

입력 alias input_dataset을 Spark dataframe으로 지정하고 colA, colB가 있는 표를 전달합니다.

def write_dataset(input_dataset):
    output = Transforms.get_output()
    output.write_dataframe(
        input_dataset,
        partition_cols=["colA", "colB"],
        output_format="csv",
    )
    return output

두 열의 값으로 파일을 분할하여 씁니다. 하나의 CSV 파일을 만드는 예제는 아닙니다.

R 파일 입력

Inputs에서 R transform input을 선택하면 fileSystem()을 사용할 수 있습니다. 스키마 없는 데이터셋은 기본적으로 이 유형으로 읽습니다. 상세 메서드와 disk_optimal은 R 파일시스템 API를 따릅니다.

RDS 읽기

입력 RDS_dataset에 R data.frame을 저장한 test_loading_RDS.rds 파일이 있다고 가정합니다.

RDS_reader <- function(RDS_dataset) {
    fs <- RDS_dataset$fileSystem()
    path <- fs$get_path("test_loading_RDS.rds", "r")
    readRDS(path)
}

ZIP의 CSV 합치기

다음 입력 zip_file_with_csvs에는 ZIP 파일이 하나 있고, 안의 CSV들은 같은 열 구조를 가진다고 가정합니다. 전체 결과는 로컬 R 메모리에 들어가야 합니다.

result <- function(zip_file_with_csvs) {
    fs <- zip_file_with_csvs$fileSystem()
    zip_name <- fs$ls()[[1]]$path
    path <- fs$get_path(zip_name, "r")
    names <- utils::unzip(path, list = TRUE)$Name
    csv_names <- names[grepl("\\.csv$", names, ignore.case = TRUE)]
    if (length(csv_names) == 0L) {
        stop("ZIP에 CSV 파일이 없습니다.")
    }

    read_member <- function(name) {
        connection <- unz(path, name, open = "r")
        on.exit(close(connection), add = TRUE)
        read.csv(connection, header = TRUE, sep = ",")
    }

    frames <- lapply(csv_names, read_member)
    do.call(rbind, frames)
}

압축 내부 파일을 임의 경로에 풀지 않고 connection으로 읽습니다. 각 connection은 사용 후 닫습니다.

R 출력 핸들

new.output()은 현재 노드의 TransformOutput을 만듭니다. 역시 데이터셋으로 저장하는 노드에서만 사용할 수 있고 콘솔에서는 파일을 쓸 수 없습니다.

R data.frame을 RDS로 저장

write_rds_file <- function(r_dataframe) {
    output <- new.output()
    fs <- output$fileSystem()
    saveRDS(r_dataframe, fs$get_path("my_RDS_file.rds", "w"))
    invisible(NULL)
}

그래프를 PDF 파일로 저장

환경에 ggplot2가 있어야 합니다. 패키지의 midwest 예제 데이터로 면적과 인구의 산점도를 만듭니다.

plot_pdf <- function() {
    data("midwest", package = "ggplot2")
    plot <- ggplot2::ggplot(midwest, ggplot2::aes(x = area, y = poptotal)) +
        ggplot2::geom_point(ggplot2::aes(colour = state, size = popdensity)) +
        ggplot2::geom_smooth(method = "loess", se = FALSE) +
        ggplot2::xlim(c(0, 0.1)) +
        ggplot2::ylim(c(0, 500000)) +
        ggplot2::theme_bw() +
        ggplot2::labs(
            subtitle = "Area Vs Population",
            y = "Population",
            x = "Area",
            title = "Scatterplot",
            caption = "Source: midwest"
        )

    output <- new.output()
    fs <- output$fileSystem()
    grDevices::pdf(fs$get_path("my pdf example.pdf", "w"))
    device <- grDevices::dev.cur()
    on.exit(grDevices::dev.off(device), add = TRUE)
    print(plot)
    invisible(NULL)
}

PDF device를 닫아 파일 기록을 마칩니다. Window나 다른 열린 그래픽 device를 모두 닫는 대신 이 함수가 연 device만 닫습니다.

텍스트 connection으로 쓰기

write_txt_file <- function() {
    output <- new.output()
    fs <- output$fileSystem()
    connection <- fs$open("my file.txt", "w")
    on.exit(close(connection), add = TRUE)
    writeLines(c("Hello", "world"), connection)
    invisible(NULL)
}

로컬 파일을 출력 경로에 업로드

다음은 로컬 임시 파일을 만든 뒤 출력 데이터셋의 output_test.txt로 올립니다.

upload <- function() {
    output <- new.output()
    fs <- output$fileSystem()
    local_path <- tempfile(fileext = ".txt")
    on.exit(unlink(local_path), add = TRUE)
    writeLines("Header 1", local_path)
    fs$upload(local_path, "output_test.txt")
    invisible(NULL)
}

Spark DataFrame의 열 분할 출력

입력 spark_df는 colA, colB가 있는 SparkR DataFrame으로 읽습니다.

write_partitioned_df <- function(spark_df) {
    output <- new.output()
    output$write.spark.df(spark_df, partition_cols = list("colA", "colB"))
    invisible(NULL)
}

명시적 출력 핸들로 저장한 파일과 일반 반환 DataFrame의 처리 방식은 구분해야 합니다. 파일 결과를 후속에서 읽으려면 저장된 데이터셋을 입력으로 연결합니다.