이전 방식의 외부 연결 변환

외부 통신 정책, 인증 정보와 반출 보안 표지를 코드에 직접 참조하는 외부 연결 변환은 Legacy 단계입니다. 추가 개발은 예정되어 있지 않지만 전체 지원은 유지됩니다. 신규 구현에는 소스 기반 외부 연결 변환을 권장합니다. 소스 기반 방식은 사설망 연결, 코드 변경 없는 인증 정보 회전과 가상 테이블 사용을 지원합니다.

이전 방식은 Code Repositories의 Python 변환에서 외부 API나 SDK를 사용하여 예약 수집 및 내보내기를 수행할 때 적용합니다. 같은 변환에 @use_external_systems()와 소스 기반 @external_systems()를 함께 붙일 수 없습니다. @use_external_systems()를 사용하는 변환은 가상 테이블과 함께 사용할 수 없으므로 소스 기반 방식으로 이전하거나 변환을 분리합니다.

저장소 설정

저장소는 SECURE 모드여야 합니다. 다른 보안 모드는 Palantir 지원과 협의해야 합니다. Foundry 입력이 없는 외부 연결 변환만 Code Repositories의 Preview helper로 미리볼 수 있습니다.

플랫폼 인스턴스의 Information Security Officer가 저장소 Settings > Repository > External systems에서 Allow access to external systems from this repository를 켭니다. 역할은 Control Panel의 Enrollment permissions에서 관리합니다.

Foundry 데이터셋을 입력으로 사용할 때에는 ISO가 해당 입력의 보안 표지와 조직을 허용 목록에 추가합니다. 예를 들어 가상 Palantir 조직의 Sensitive 데이터라면 두 항목이 모두 필요합니다. 설정의 Configure use of Foundry inputs with external systems 또는 Allow use of Foundry inputs with external systems in this repository에서 구성합니다.

Libraries에서 transforms-external-systems를 추가합니다. VS Code 작업 공간에서는 위쪽 Settings > External systems에서 Egress와 Credentials 설정에 접근합니다.

외부 통신 정책 가져오기

사용자 코드가 외부로 시작하는 연결은 외부 통신 정책에 연결되어야 합니다. 셀프서비스가 활성화된 플랫폼 인스턴스는 Control Panel에서 정책을 요청하고 관리할 수 있습니다. 새 경로나 기능 활성화가 필요하면 ISO 또는 Palantir 지원에 요청합니다.

전역 정책도 저장소의 Egress and Credentials > Egress에서 가져와야 합니다.

원천이 접속 IP를 제한한다면 Foundry의 실제 발신 IP도 허용되어야 합니다. 필요한 주소는 Palantir 지원을 통해 확인합니다.

인증 정보와 인증서

Credentials 탭에서 외부 시스템의 사용자 이름과 비밀번호 등 필요한 값을 등록합니다. 공개 API처럼 인증 정보가 필요 없는 경우에는 Credential 입력을 선언하지 않아도 됩니다. TLS 인증서가 필요한 연결은 인증 정보를 저장하는 리소스에 인증서를 보관하고 코드에서 해당 클라이언트에 전달할 수 있습니다.

코드에서는 use_external_systems, EgressPolicy, 필요한 경우 Credential을 가져옵니다. 데코레이터에 선언한 이름과 compute 함수의 인수 이름을 맞춥니다.

실제 연결 값을 받는 변환 생성 함수

다음 예제 함수들은 실제 정책 RID, 인증 정보 RID와 출력 위치를 인수로 받아 완성된 Transform 객체를 반환합니다. 반환한 객체를 프로젝트의 Pipeline에 add_transforms()로 등록합니다. RID를 추정하거나 인증 정보 값을 소스 코드에 넣지 않습니다.

인증된 API 응답 수집

api_url은 해당 외부 통신 정책이 허용하는 실제 JSON API 주소입니다. 인증 정보에는 username과 password 키가 있어야 합니다.

import requests
from transforms.api import Output, transform
from transforms.external.systems import Credential, EgressPolicy, use_external_systems


def make_api_download(policy_rid: str, credential_rid: str, output_path: str, api_url: str):
    @use_external_systems(
        egress=EgressPolicy(policy_rid),
        creds=Credential(credential_rid),
    )
    @transform(output=Output(output_path))
    def compute(egress, creds, output):
        response = requests.get(
            api_url,
            auth=(creds.get("username"), creds.get("password")),
            timeout=10,
        )
        response.raise_for_status()
        with output.filesystem().open("response.json", "w") as destination:
            destination.write(response.text)

    return compute

여러 정책과 인증 정보

하나의 변환에 여러 정책 및 Credential을 별도 이름으로 선언할 수 있습니다. 다음 함수는 두 실제 JSON API를 각각의 인증 정보로 조회하여 실제 응답을 한 JSON 파일에 기록합니다.

import json
import requests
from transforms.api import Output, transform
from transforms.external.systems import Credential, EgressPolicy, use_external_systems


def make_two_api_download(
    first_policy_rid: str,
    second_policy_rid: str,
    first_credential_rid: str,
    second_credential_rid: str,
    output_path: str,
    first_url: str,
    second_url: str,
):
    @use_external_systems(
        egress1=EgressPolicy(first_policy_rid),
        egress2=EgressPolicy(second_policy_rid),
        creds1=Credential(first_credential_rid),
        creds2=Credential(second_credential_rid),
    )
    @transform(output=Output(output_path))
    def compute(egress1, egress2, creds1, creds2, output):
        responses = {}
        for label, url, credentials in [
            ("first", first_url, creds1),
            ("second", second_url, creds2),
        ]:
            response = requests.get(
                url,
                auth=(credentials.get("username"), credentials.get("password")),
                timeout=10,
            )
            response.raise_for_status()
            responses[label] = response.json()
        with output.filesystem().open("responses.json", "w") as destination:
            json.dump(responses, destination)

    return compute

Foundry 입력과 ExportControl

Foundry 데이터와 외부 연결을 같은 작업에서 사용하면 실제 내보내기 여부와 관계없이 데이터가 외부로 나갈 가능성이 있습니다. 개발자는 외부 시스템에서의 보호 조건을 검토해야 합니다.

ExportControl에는 반출할 보안 표지 ID와 조직 ID를 함께 넣습니다. 업스트림 데이터에 추가 보안 표지가 있으면 작업이 실패합니다. 오류의 missing markings가 보안 표지가 아니라 조직을 가리킬 수도 있습니다. 허용 범위를 늘리려면 ISO가 저장소 설정도 변경해야 합니다.

다음 예제는 입력 데이터셋의 실제 URL에서 파일을 읽습니다. Name은 확장자를 포함한 고유한 파일 이름, Url은 해당 파일의 실제 주소인 문자열 열입니다. 소규모 파일 목록을 드라이버에서 순회하며 HTTP 오류를 출력 성공으로 바꾸지 않습니다.

import requests
from transforms.api import Input, Output, transform
from transforms.external.systems import EgressPolicy, ExportControl, use_external_systems


def make_file_download(
    policy_rid: str,
    allowed_marking_ids: list[str],
    input_path: str,
    output_path: str,
):
    @use_external_systems(
        export_control=ExportControl(markings=allowed_marking_ids),
        egress=EgressPolicy(policy_rid),
    )
    @transform(image_urls=Input(input_path), images_output=Output(output_path))
    def compute(export_control, egress, image_urls, images_output):
        filesystem = images_output.filesystem()
        for row in image_urls.dataframe().select("Name", "Url").toLocalIterator():
            with requests.get(row.Url, stream=True, timeout=30) as response:
                response.raise_for_status()
                with filesystem.open(row.Name, "wb") as destination:
                    for chunk in response.iter_content(chunk_size=65536):
                        destination.write(chunk)

    return compute

파일 경로와 바이트를 다루는 방식은 일반 변환의 파일 인터페이스와 같습니다. 병렬화하는 경우에는 실행 재시도와 중복 파일 이름의 영향을 별도로 고려해야 합니다.

미디어 집합 출력

이미지, 음성이나 PDF 같은 파일을 미디어 집합에 넣을 때는 transforms-media 라이브러리의 MediaSetOutput을 사용합니다. 대상 미디어 집합의 스키마는 실제 파일과 맞아야 합니다.

다음 함수는 정책이 허용한 실제 URL의 바이트를 임시 파일에 스트리밍한 뒤, 지정한 항목 경로에 기록합니다. 인증이 필요하지 않은 외부 URL을 받는 예제입니다.

import tempfile
import requests
from transforms.api import transform
from transforms.mediasets import MediaSetOutput
from transforms.external.systems import EgressPolicy, use_external_systems


def make_media_download(
    policy_rid: str,
    media_set_path: str,
    file_url: str,
    media_item_path: str,
):
    @use_external_systems(egress=EgressPolicy(policy_rid))
    @transform(media_set_output=MediaSetOutput(media_set_path))
    def compute(egress, media_set_output):
        with requests.get(file_url, stream=True, timeout=30) as response:
            response.raise_for_status()
            with tempfile.TemporaryFile() as temporary_file:
                for chunk in response.iter_content(chunk_size=65536):
                    temporary_file.write(chunk)
                temporary_file.seek(0)
                media_set_output.put_media_item(temporary_file, path=media_item_path)

    return compute

외부 SDK 사용: Mapbox Static

외부 시스템의 Python SDK도 사용할 수 있습니다. 다음은 mapbox 패키지의 Static Maps V4용 Static 클라이언트로 런던 주변 좌표의 지도를 받아 저장하는 예제입니다. 이 API에 접근할 수 있는 실제 Mapbox 토큰을 Credential의 mapbox-token 키에 저장하고, Mapbox API 도메인의 외부 통신 정책을 준비합니다.

from mapbox import Static
from transforms.api import Output, transform
from transforms.external.systems import Credential, EgressPolicy, use_external_systems


def make_mapbox_image(policy_rid: str, credential_rid: str, output_path: str):
    @use_external_systems(
        egress=EgressPolicy(policy_rid),
        mapbox_creds=Credential(credential_rid),
    )
    @transform(output=Output(output_path))
    def compute(egress, mapbox_creds, output):
        client = Static(access_token=mapbox_creds.get("mapbox-token"))
        response = client.image("mapbox.satellite", lat=51.5072, lon=0.0, z=12)
        if response.status_code != 200:
            raise RuntimeError(f"Mapbox 이미지 요청 HTTP {response.status_code}")
        with output.filesystem().open("london.log", "w") as log_file:
            log_file.write(f"response code: {response.status_code}")
        with output.filesystem().open("london.png", "wb") as image_file:
            image_file.write(response.content)

    return compute

토큰이 포함될 수 있는 요청 URL은 로그에 남기지 않습니다. SDK의 실제 응답 바이트를 이미지로 저장합니다. SDK 릴리스와 외부 서비스의 지원 범위는 별도로 확인합니다.

증분 처리와 외부 상태

일반 증분 변환은 Foundry 입력 파일과 이전 처리 트랜잭션을 비교합니다. 외부 시스템의 커서나 조회 위치는 개발자가 명시적으로 보관해야 합니다. 숨김 파일 _state.json처럼 이름을 밑줄로 시작하면 데이터셋 표 미리보기에서 상태 파일을 제외할 수 있습니다.

다음 생성 함수는 실제 조회 함수 fetch_batch를 필수 인수로 받습니다. 조회 함수의 계약은 다음과 같습니다.

• 입력은 변환 컨텍스트, Credential로 선언하여 실행 시 전달받은 인증 정보, 이전 상태 딕셔너리입니다.
• 반환값은 실제 외부 자료로 만든 PySpark DataFrame과 다음 상태 딕셔너리의 튜플입니다.
• 상태는 JSON으로 저장할 수 있어야 합니다. 조회 실패는 예외로 전달해야 합니다.
• 초기 상태는 원천의 실제 커서 계약에 맞춰 호출자가 제공합니다. 예를 들어 숫자 커서를 쓰는 시스템에서 {"last_seen": 0}을 사용할 수 있지만 모든 API의 시작 상태는 아닙니다.

이 인터페이스에 프로젝트의 실제 원천 조회 함수를 전달합니다.

import json
from collections.abc import Callable
from typing import Any
from pyspark.sql import DataFrame
from transforms.api import Output, configure, incremental, transform
from transforms.external.systems import Credential, EgressPolicy, use_external_systems


def make_incremental_export(
    policy_rid: str,
    credential_rid: str,
    output_path: str,
    initial_state: dict[str, Any],
    fetch_batch: Callable[[Any, Any, dict[str, Any]], tuple[DataFrame, dict[str, Any]]],
):
    @incremental()
    @configure(profile=["KUBERNETES_NO_EXECUTORS"])
    @use_external_systems(
        egress=EgressPolicy(policy_rid),
        creds=Credential(credential_rid),
    )
    @transform(out=Output(output_path))
    def compute(ctx, out, egress, creds):
        if ctx.is_incremental:
            with out.filesystem(mode="previous").open("_state.json", "r") as state_file:
                state = json.load(state_file)
        else:
            state = initial_state

        frame, next_state = fetch_batch(ctx, creds, state)
        out.set_mode("modify" if ctx.is_incremental else "replace")
        out.write_dataframe(frame)
        with out.filesystem(mode="current").open("_state.json", "w") as state_file:
            json.dump(next_state, state_file)

    return compute

상태 파일이 있어야 하는 증분 실행에서 파일이 없거나 손상되면 예외를 전달합니다. 모든 오류를 잡아서 초기 상태로 돌아가면 중복 수집이나 누락을 감출 수 있습니다. 상태는 데이터 쓰기가 끝난 뒤 같은 출력에 기록하며, 변환이 실패하면 이를 성공한 다음 커서로 취급하지 않습니다. 외부 시스템에 쓰는 작업까지 Foundry 출력과 하나의 트랜잭션으로 묶이는 것은 아닙니다.

기존 정책, 인증 정보와 ExportControl을 Data Connection 소스로 이전하는 절차와 소스 설정이 반영되는 빌드 경계는 소스 기반 변환에 있습니다.