R TransformInput, TransformOutput과 파일시스템

R transform input은 데이터셋의 표와 파일에 접근하는 저수준 입력 핸들입니다. 노드의 Inputs에서 R transform input으로 지정해야 합니다. 출력은 데이터셋으로 저장하는 R 노드에서 new.output()으로 만듭니다.

TransformInput

R 객체의 메서드는 $로 호출합니다.

예를 들어 input_dataset$fileSystem()으로 파일 목록과 읽기 경로를 얻습니다.

TransformOutput

write.spark.df(df, partition_cols=NULL, bucket_cols=NULL, bucket_count=NULL, sort_by=NULL)

Spark DataFrame을 출력 데이터셋에 씁니다.

sort_by는 버킷 파일을 쓸 때의 정렬 설정이며 이후 모든 데이터셋 조회의 순서를 보장하는 옵션은 아닙니다.

write.data.frame(rdf)

로컬 R data.frame을 출력 데이터셋에 씁니다. 인자 rdf에는 실제 표를 전달합니다.

fileSystem()

출력 파일시스템을 반환합니다. 텍스트, RDS, PDF 등 비정형 파일을 논리 경로에 쓸 수 있습니다. 구체적인 입출력 예제는 저장 노드의 권한과 실행 조건을 따릅니다.

FileSystem의 파일 목록

ls(glob=NULL, regex='.*', show_hidden=FALSE)

데이터셋 루트를 기준으로 패턴에 맞는 파일을 반환합니다.

결과의 각 FileStatus 항목에는 path, size, modified가 있습니다. path는 데이터셋 안의 논리 경로, size는 바이트 수, modified는 1970년 1월 1일 UTC부터의 밀리초 시각입니다. 코드에서는 목록의 각 항목을 [[i]]로 선택하고 $path 등으로 필드를 읽을 수 있습니다.

Connection과 로컬 경로

open(path, open='r', disk_optimal=FALSE, encoding=default)

Foundry 파일을 열어 R connection을 반환합니다.

읽기에는 r, 쓰기에는 w, 바이너리에는 해당 라이브러리가 요구하는 rb나 wb를 사용합니다. 실제 connection이 지원하는 모드여야 하며 사용 후 닫아 버퍼와 자원을 정리합니다.

get_path(path, open='r', disk_optimal=FALSE, encoding=default)

원격 논리 경로에 대응하는 로컬 임시 경로 문자열을 반환합니다. 인자들의 의미와 기본값은 open()과 같습니다. 파일 경로를 요구하는 readRDS()나 그래픽 device 등에 전달할 수 있습니다.

반환 경로는 데이터셋의 원래 저장 위치가 아니라 실행 환경에서 사용할 임시 경로입니다.

로컬 파일 업로드

upload(local_path, remote_path)

로컬 파일을 출력 파일시스템의 논리 경로로 올립니다. 쓰기 전용 작업이며 별도의 데이터 결과를 반환하는 연산은 아닙니다.

disk_optimal의 읽기 방식

기본값 FALSE에서는 접근하기 전에 파일을 내려받습니다. Seek 가능한 일반 파일을 요구하는 라이브러리에 적합합니다.

TRUE에서는 코드가 실행되는 동안 파일을 내려받는 스트리밍 경로를 사용합니다. get_path()로 받은 경로를 직접 열 때는 fifo()로 읽어야 합니다. 모든 라이브러리가 FIFO를 지원하는 것은 아니므로 일반 파일 경로와 동일하게 취급하지 않습니다. 큰 파일의 일부만 읽을 때 사용할 수 있습니다.

아래는 입력 alias large_txt_file을 R transform input으로 지정하고 그 안의 large_txt_file.txt에서 처음 열 줄을 읽어 출력하는 노드입니다.

disk_optimal_example <- function(large_txt_file) {
    fs <- large_txt_file$fileSystem()
    connection <- fs$open(
        "large_txt_file.txt", "r", disk_optimal = TRUE
    )
    on.exit(close(connection), add = TRUE)
    lines <- readLines(connection, n = 10)
    print(lines)
    NULL
}

이 노드는 콘솔 출력만 남기며 NULL을 하위 데이터 입력으로 사용할 수 없습니다. 출력 파일시스템에 쓴 파일을 같은 처리에서 다시 읽어야 하면 disk_optimal=FALSE를 사용해야 합니다.