컬렉션, JSON과 체크포인트
이 장의 열 함수는 from pyspark.sql import functions as F로 가져옵니다. DataFrame의 checkpoint 메서드와 열 함수는 호출 대상을 구분합니다.
배열과 구조체
| 함수 | 동작 |
|---|---|
F.array(*cols) | 여러 열 값을 배열 열로 만듦 |
F.array_contains(col, value) | 배열에 지정 값이 있는지 검사 |
F.size(col) | 컬렉션 크기 |
F.sort_array(col, asc=True) | 배열 원소 정렬, 기본 오름차순 |
F.struct(*cols) | 여러 열을 구조체 열로 묶음 |
F.create_map(*cols) | 키/값 열을 번갈아 전달하여 map 열 생성 |
F.explode(col) | 배열이나 map의 항목을 여러 행으로 펼침 |
F.posexplode(col) | 항목과 위치를 함께 여러 행으로 펼침 |
배열/구조체를 만든다는 것은 Python list나 dict로 전체 데이터를 수집한다는 뜻이 아닙니다. 결과는 각 행 안에 중첩 값을 담는 Spark 열입니다.
열 선택과 정렬 표현식
F.col(name)과 F.column(name)은 이름으로 열 표현식을 만듭니다. F.expr(sql_expression)은 SQL 표현식 문자열을 열 표현식으로 파싱합니다.
F.asc(col)과 F.desc(col)는 정렬 방향 표현식입니다. DataFrame의 sort/orderBy에 연결해야 전체 행을 정렬합니다.
from pyspark.sql import functions as F
def nested_contact(df):
return df.select(
"id",
F.struct("name", "email").alias("contact"),
F.array("home_phone", "work_phone").alias("phones"),
)
예제 입력에는 id, name, email, home_phone과 work_phone 열이 있어야 합니다.
비트 연산
| 현재 이름 | 기존 이름 | 동작 |
|---|---|---|
F.bitwise_not(col) | F.bitwiseNOT(col) | 비트 반전 |
F.shiftleft(col, numBits) | F.shiftLeft(col, numBits) | 왼쪽 비트 이동 |
F.shiftright(col, numBits) | F.shiftRight(col, numBits) | 부호를 유지하는 오른쪽 이동 |
F.shiftrightunsigned(col, numBits) | F.shiftRightUnsigned(col, numBits) | 부호 없는 오른쪽 이동 |
기존 이름을 사용하는 코드도 볼 수 있지만 새 코드는 현재 이름으로 작성합니다.
NULL과 NaN
| 함수 | 의미 |
|---|---|
F.coalesce(*cols) | 순서대로 첫 번째 NULL이 아닌 값 선택 |
F.isnan(col) | NaN 여부 |
F.isnull(col) | NULL 여부 |
NaN과 NULL은 다른 값입니다. coalesce는 NaN을 자동으로 건너뛰는 함수가 아닙니다. 또한 F.coalesce()는 값을 선택하는 열 함수이며 파티션을 줄이는 df.coalesce(n)과 다릅니다.
해시와 입력 정보
| 함수 | 의미 |
|---|---|
F.hash(*cols) | 여러 열의 해시 값 |
F.sha1(col) | SHA-1 해시 표현 |
F.sha2(col, numBits) | 지정 비트 길이의 SHA-2 해시 표현 |
F.soundex(col) | 문자열의 SoundEx 인코딩 |
F.input_file_name() | 현재 파일 읽기 작업의 입력 파일 이름 열 |
F.spark_partition_id() | 현재 Spark 파티션 ID 열 |
파티션 ID는 실행 계획과 분할에 따른 값이며 업무 레코드의 영구 식별자가 아닙니다. 해시를 적용했다는 사실만으로 민감 데이터의 익명성이 보장되지는 않습니다.
JSON 파싱과 생성
| 함수 | 결과와 역할 |
|---|---|
F.from_json(col, schema, options) | JSON 문자열을 지정 schema의 중첩 값으로 파싱. options는 생략 가능한 옵션 맵 |
F.get_json_object(col, path) | JSON 경로에 해당하는 값을 문자열로 추출 |
F.json_tuple(col, *fields) | 지정한 필드를 여러 문자열 열로 추출 |
F.to_json(col, options) | struct/array/map 같은 중첩 값을 JSON 문자열로 생성. options 생략 가능 |
from pyspark.sql import functions as F
from pyspark.sql import types as T
PAYLOAD_SCHEMA = T.StructType([
T.StructField("id", T.IntegerType(), True),
T.StructField("name", T.StringType(), True),
])
def parse_payloads(df):
return df.withColumn("parsed", F.from_json("payload", PAYLOAD_SCHEMA)).select(
F.col("parsed.id").alias("id"),
F.col("parsed.name").alias("name"),
F.to_json("parsed").alias("normalized_json"),
)
예제 입력의 payload는 JSON 문자열 열입니다. 실제 사용에서는 입력의 자료형과 파서 옵션을 원하는 실패 처리에 맞춰 지정합니다.
checkpoint와 localCheckpoint
두 메서드는 긴 논리 계획을 끊어 새로운 DataFrame을 만들 때 사용합니다.
| 메서드 | 저장과 신뢰성 |
|---|---|
df.checkpoint(eager=True) | 설정한 checkpoint 파일시스템에 저장. SparkContext에 checkpoint 디렉터리 설정 필요 |
df.localCheckpoint(eager=True) | executor의 캐시 저장 체계 사용. executor 손실에 대한 신뢰 가능한 복구 지점이 아님 |
기본 eager=True는 즉시 체크포인트를 만듭니다. False이면 실제 계산 시점을 뒤로 미룹니다. localCheckpoint는 Experimental API입니다.
def checkpoint_with_shared_directory(df, checkpoint_directory):
df.sparkSession.sparkContext.setCheckpointDir(checkpoint_directory)
return df.checkpoint(eager=True)
def local_checkpoint(df):
return df.localCheckpoint(eager=True)
첫 함수의 경로는 cluster가 접근할 수 있도록 준비한 실제 공유 파일시스템 경로여야 합니다. 임의의 driver 로컬 경로를 모든 executor가 공유한다고 가정하지 않습니다. 이 체크포인트는 Foundry 출력 데이터셋과 빌드 이력을 등록하는 write_dataframe()을 대신하지 않습니다.