파일을 주고받는 CLI 사이드카

CLI 사이드카는 입력 파일을 공유 볼륨에 놓고, 컨테이너 안에서 명령을 실행한 뒤 결과 파일을 회수합니다. 예제 명령은 dd if=infile.csv of=outfile.csv입니다. CSV를 그대로 복사하는 실제 명령을 사용해 파일 준비, 실행 완료, 오류 전달과 결과 회수 절차를 구성합니다.

명령은 작업 디렉터리의 파일을 읽고 결과 파일을 쓴 뒤 종료해야 합니다. 지속적으로 요청을 받는 서버 프로그램에는 HTTP 사이드카를 사용합니다.

파일 준비

myproject는 예제의 Python 패키지 이름입니다. 저장소에서 다른 패키지 이름을 사용하면 파일 배치와 두 변환의 import 경로를 함께 바꿉니다. 입력과 출력에 지정한 /training/containers/... 경로도 자신의 프로젝트에서 읽거나 쓸 수 있는 실제 경로로 설정합니다.

예제 CSV에는 data1, data2, data3 열과 네 행이 있습니다. 단일 실행에서는 데이터셋 루트에 infile.csv라는 파일이 있어야 합니다. 병렬 실행에서는 같은 열 이름을 가진 테이블로 읽을 수 있도록 입력 데이터셋의 스키마와 파일 읽기 설정을 구성합니다.

컨테이너와 공유 볼륨

Dockerfile은 Python 3.12 slim 이미지에 coreutils를 설치하고 UID 5001로 실행합니다. Python은 요청 제어 프로그램에 필요하며, dd는 입력 파일을 복사합니다. /opt/palantir/sidecars/shared-volumes/shared 디렉터리를 만들고 실행 UID가 쓸 수 있게 설정합니다.

이미지의 진입점은 명령과 각 인자를 JSON 배열로 전달합니다.

ENTRYPOINT ["python", "-u", "/app/entrypoint.py", "--", "dd", "if=infile.csv", "of=outfile.csv"]

entrypoint.py는 명령을 셸 문자열로 해석하지 않습니다. 프로그램을 바꿀 때에는 실행 파일과 각 인자를 배열의 별도 원소로 지정합니다. 파일명에 공백이 있더라도 해당 인자 전체를 한 원소로 넣습니다.

Artifacts 저장소 인증을 마친 뒤, 실제 이미지 경로를 FOUNDRY_CONTAINER_REGISTRY에 설정하고 빌드 폴더에서 다음 명령을 사용합니다.

: "${FOUNDRY_CONTAINER_REGISTRY:?Artifacts 저장소에 표시된 이미지 경로를 설정하세요}"
IMAGE="${FOUNDRY_CONTAINER_REGISTRY}/simple-cli:0.0.1"
docker build --platform linux/amd64 --tag "$IMAGE" .
docker push "$IMAGE"

코드 저장소에는 transforms-sidecar 라이브러리와 Docker Artifacts 저장소를 추가합니다. 두 변환 모두 다음 선언으로 각 실행기에 같은 이미지의 사이드카와 공유 볼륨을 연결합니다.

@sidecar(image="simple-cli", tag="0.0.1", volumes=[Volume("shared")])

이미지에 디렉터리를 만드는 것과 Volume("shared")를 선언하는 작업이 모두 필요합니다. 디렉터리만 만들면 실행기와 사이드카 사이에 공유 볼륨이 연결되지 않습니다.

한 번의 작업으로 전체 입력 파일 처리

single_transform.py는 파티션 하나에 작업 하나를 배치합니다. 작업 함수는 실행기에서 다음 순서로 동작합니다.

1. 입력 데이터셋의 파일을 새 요청의 data 디렉터리에 복사합니다. 중첩된 상대 경로는 유지합니다.
2. 모든 입력 파일을 닫은 뒤 start_flag를 생성합니다.
3. 사이드카가 dd를 실행하고 완료 상태를 기록할 때까지 기다립니다.
4. 반환 코드와 시간 제한 초과 여부를 확인합니다.
5. start_flag, outfile.csv, logfile, done_flag를 출력 데이터셋에 복사합니다.
6. close_flag로 결과 회수가 끝났음을 알립니다.

dd는 요청의 data 디렉터리를 작업 디렉터리로 사용하므로, infile.csv는 입력 데이터셋 루트에 있어야 합니다. 다른 이름으로 올렸거나 하위 폴더에만 있으면 명령이 실패하고 변환에도 오류가 전달됩니다.

작업 배치는 parallelize([1], 1).foreach(run_once)입니다. 논리적으로 작업 하나를 실행기에 보내는 방식이며, Spark의 작업 재시도까지 금지한다는 뜻은 아닙니다. foreach에는 반환값이 필요하지 않습니다. 작업이 완료되었다는 판단은 실제 명령 결과와 파일 복사 완료에 근거합니다.

여러 입력 행을 병렬 처리

parallel_transform.py는 각 입력 행을 한 행짜리 CSV로 저장한 뒤 같은 CLI를 실행합니다. Spark 실행기 프로필은 다음 두 가지를 사용하므로, 저장소의 Settings > Spark에서 해당 프로필을 가져옵니다.

@configure(profile=["EXECUTOR_CORES_EXTRA_SMALL", "NUM_EXECUTORS_4"])

입력은 repartition(4)로 네 파티션에 분배합니다. 이 설정은 실행기의 작업 슬롯과 실행기 수를 구성하지만, 네 입력 행이 반드시 서로 다른 실행기 하나씩에 배정된다고 가정하지 않습니다. 실행기가 여러 요청을 순차 처리할 수 있도록 사이드카의 진입점은 계속 실행됩니다.

각 행의 처리 절차는 다음과 같습니다.

• 전체 UUID를 포함한 새 요청 디렉터리를 만듭니다. 입력과 플래그의 이름이 다른 요청과 충돌하지 않습니다.
• UTF-8 CSV에 data1,data2,data3 헤더와 해당 행을 씁니다. Python CSV writer와 reader는 같은 기본 인용 규칙을 사용합니다.
• 완료 상태를 확인한 후 결과 CSV의 열 이름이 정확하고 데이터 행이 하나인지 검사합니다. 누락된 값이나 열 개수 불일치도 오류로 처리합니다.
• 다섯 파일 start_flag, infile.csv, outfile.csv, logfile, done_flag를 원시 파일 출력 데이터셋에 복사합니다. 각 파일명에는 요청 UUID가 들어간 접두사를 붙입니다.
• 원래 세 열, success, 결과의 outdata1, outdata2, outdata3를 테이블 출력으로 반환합니다.

success의 값은 명령 반환 코드가 0이고, 오류나 시간 제한 초과가 없으며, 결과 CSV와 파일 복사까지 정상 완료된 뒤에만 "success"로 설정합니다. 실패한 행을 빈 문자열 결과나 성공 행으로 바꾸지 않습니다.

원래 세 열은 입력 스키마의 자료형과 nullable 설정을 유지합니다. CSV에서 다시 읽은 세 결과 열은 문자열입니다. csv.writer가 None을 빈 필드로 기록하므로 CSV 쪽에서는 NULL과 빈 문자열을 구분하지 않습니다. 원래 입력 열은 그 값과 자료형을 그대로 보존합니다. 출력 스키마를 명시하므로 입력이 비어 있어도 열 구조를 추론할 필요가 없습니다.

반복 실행과 오류 처리

공유 볼륨에는 요청별 디렉터리가 있으며, 사이드카는 요청 하나를 처리한 뒤 결과 회수 신호를 기다리고 해당 디렉터리만 정리합니다. 이어서 다음 요청을 처리합니다. Spark 실행기가 종료될 때까지 컨테이너를 유지하는 구성과, 요청 하나의 완료를 구분합니다.

명령 실행, 클라이언트의 완료 대기, 결과 회수 신호 대기에는 각각 600초 설정을 사용합니다. 클라이언트 대기에는 이미 시작 신호를 보낸 요청이 앞선 작업을 기다리는 시간도 포함됩니다. 실제 작업 시간과 결과 파일 크기에 맞춰 서버와 클라이언트 설정을 함께 조정해야 합니다.

Spark 재시도나 같은 계산의 반복 실행은 CLI를 다시 호출할 수 있습니다. 요청 UUID는 파일 충돌을 막지만, 외부 상태 변경을 한 번만 수행하도록 보장하지는 않습니다. 예제의 dd는 요청 전용 로컬 파일만 복사합니다. 외부 시스템을 변경하는 명령으로 교체하면 해당 시스템의 멱등성이나 중복 실행 처리 규약이 필요합니다.

공유 볼륨 요청 규약에는 상태 파일, 오류 전달, 각 유틸리티 함수와 프로세스 종료 조건이 정리되어 있습니다.