Keyboard shortcuts

Press ← or → to navigate between chapters

Press ? to show this help

Press Esc to hide this help

소스 기반 외부 연결 변환

외부 연결 변환(External transforms)은 Python 변환 저장소에서 외부 시스템에 연결하여 데이터를 읽거나 쓰는 방식입니다. 필요한 커넥터가 없거나, 소스의 수집 및 내보내기 기능이 원하는 동작을 제공하지 않을 때 사용합니다. 사용자 지정 쿼리, 수집 중 파일 묶기, 전송 중 암호화 또는 복호화 같은 처리를 코드에 넣을 수 있습니다.

현재 권장 방식은 Data Connection 소스를 저장소에 직접 가져오는 것입니다. 정책과 인증 정보를 코드에 직접 연결하는 이전 방식도 지원 범위 안에서 운영할 수 있지만 새 구현은 소스 기반으로 구성합니다. 외부 인증 애플리케이션(Outbound application)으로 인증하는 소스는 외부 연결 변환에서 지원하지 않습니다.

가상 테이블을 사용하는 변환도 외부 데이터가 있는 시스템에 연결해야 하므로 외부 연결 변환에 해당합니다. 해당 소스의 코드 가져오기와 반출 설정이 필요합니다.

저장소와 소스 준비

Python transforms 저장소를 만듭니다. 일반 Python 변환의 기능을 외부 연결 변환과 함께 사용할 수 있습니다. 예제 연결에는 공개 Dictionary API를 사용합니다. 이 서비스는 Palantir와 무관하며 동작이 바뀔 수 있으므로 운영 서비스 보증으로 해석하지 않습니다.

저장소에서 소스 생성

  1. 왼쪽 External systems를 엽니다. VS Code 작업 공간에서는 위쪽 Settings > External systems를 사용합니다.
  2. Add > Create new를 선택합니다.
  3. 소스 이름과 저장할 프로젝트를 선택합니다.

생성된 소스의 측면 패널에서 외부 통신 정책, 비밀 정보와 반출 가능한 보안 표지를 구성할 수 있습니다.

Dictionary API에는 api.dictionaryapi.dev의 HTTPS 연결 정책을 추가합니다. 이 API에는 인증 비밀 정보가 필요하지 않습니다. Foundry 입력 데이터를 쓰지 않는 수집 예제는 아직 반출 설정을 켜지 않아도 되지만, 이후 Foundry 입력을 함께 사용하려면 반출 설정이 필요합니다.

REST 연결이면 기본 requests 클라이언트를 사용할 수 있도록 Generic 소스를 REST API 소스로 전환하는 안내가 표시됩니다.

Data Connection에서 소스 생성

Data Connection의 New Source > REST API에서도 만들 수 있습니다. Dictionary API에는 Foundry 워커를 선택하고 도메인 및 외부 통신 정책을 구성합니다. 에이전트 워커 연결은 외부 연결 변환에서 지원하지 않습니다. 사설망에는 Foundry 워커와 에이전트 프록시 정책을 사용합니다.

소스를 생성한 것만으로 저장소에서 사용할 수 있는 것은 아닙니다. 다음의 코드 가져오기 허용을 완료합니다.

라이브러리와 코드 가져오기

저장소의 Libraries에서 transforms-external-systems를 찾아 설치합니다.

소스의 Connection settings > Code import configuration에서 Allow this source to be imported into code repositories를 켭니다. 이 화면에는 소스를 가져온 저장소 목록도 표시됩니다.

저장소의 External Systems > Add에서 소스를 찾아 Confirm selection으로 가져옵니다. 가져오려면 소스에 최소 Editor 권한이 필요합니다.

여러 도메인을 가진 REST API 소스는 가져올 수 없습니다. 여러 도메인이 필요하면 도메인별 소스를 만들고 같은 변환에 각각 선언합니다.

연결과 비밀 정보

transforms.external.systems의 @external_systems와 Source로 사용할 소스를 선언합니다. 실행 시 같은 이름의 인수에 연결 객체가 들어옵니다. 코드 편집기는 RID를 소스 이름과 Data Connection 링크로 표시합니다.

get_https_connection().url은 기본 URL, get_https_connection().get_client()는 소스의 인증서와 연결 설정을 반영한 requests Session을 제공합니다. 비밀 정보는 get_secret()에 실제 키를 전달하여 읽습니다. 사용할 키 이름은 소스 측면 패널에서 확인합니다.

모든 소스가 공통 hostname 같은 속성을 제공하지는 않습니다. API에서 구성을 제공하는 유형에는 source_configuration을 사용할 수 있습니다. 세션 인증, 프록시 URI와 소켓은 Python 소스 인터페이스에 설명되어 있습니다.

기본 HTTPS 클라이언트는 서버 및 클라이언트 인증서를 반영합니다. 다른 라이브러리를 사용하면 해당 클라이언트에 필요한 프록시, 소켓과 TLS 설정을 전달해야 합니다. 변환이나 UDF에서 os.chdir()를 호출하면 보안 연결의 환경 변수 참조가 깨질 수 있습니다.

고정 단어 목록을 조회하는 Pandas 변환

다음 예제의 RID와 출력 경로는 가상 값입니다. 실제 가져온 소스와 출력 위치로 변경합니다. 라이브러리에는 pandas와 transforms-external-systems가 필요합니다.

import pandas as pd
from transforms.api import Output, transform_pandas
from transforms.external.systems import Source, external_systems

DICTIONARY_SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
STATIC_OUTPUT = "/Examples/external/static-phonetics"


@external_systems(dictionary_api_source=Source(DICTIONARY_SOURCE_RID))
@transform_pandas(Output(STATIC_OUTPUT))
def fetch_static_words(dictionary_api_source) -> pd.DataFrame:
    connection = dictionary_api_source.get_https_connection()
    client = connection.get_client()
    rows = []
    for word in ["apple", "dog", "cat"]:
        response = client.get(connection.url + "/api/v2/entries/en/" + word, timeout=10)
        response.raise_for_status()
        entry = response.json()[0]
        rows.append({"word": word, "phonetic": entry.get("phonetic")})
    return pd.DataFrame.from_records(rows, columns=["word", "phonetic"]).astype(
        {"word": "string", "phonetic": "string"}
    )

응답에 발음 표기 phonetic이 없으면 해당 값은 결측값입니다. HTTP 요청 자체가 실패하면 예외를 전달합니다.

Foundry 입력과 반출 통제

Foundry 데이터셋의 값을 외부 API에 보내거나 외부 데이터와 같은 작업에서 처리하면 데이터가 외부로 나갈 가능성이 생깁니다. 실제 코드가 외부로 데이터를 쓰지 않더라도 반출 통제 대상입니다.

소스의 Connection settings > Export configuration에서 Enable exports to this source를 켜고, 반출을 허용할 보안 표지와 조직을 지정합니다. 기본 역할에서 토글을 변경하려면 소스 Owner와 플랫폼 인스턴스의 Information Security Officer가 필요합니다. 보안 표지 추가에는 소스 Owner와 표지의 Apply/Remove 권한이, 다른 조직 추가에는 소스 Owner와 그 조직의 Expand access 권한이 필요합니다. 자신의 조직에는 Expand access 예외가 있습니다. 표지나 조직 추가 자체에 ISO 역할을 다시 요구하지는 않습니다.

이 설정은 다음 경우에 필요합니다.

  • 소스를 가져온 Python 변환에서 데이터셋, 미디어 집합 또는 스트림을 입력으로 사용합니다.
  • 해당 소스에 등록된 가상 테이블을 Python 변환에서 사용합니다.

가상 예로 Palantir 조직의 데이터만 허용하려면 해당 조직을 반출 구성에 추가합니다. 별도 보안 표지가 붙은 입력은 그 표지도 허용되어야 합니다.

입력 데이터셋의 단어를 조회하는 변환

입력에는 결측값이 없는 문자열 열 word가 있어야 합니다. 다음 예제는 각 요청의 HTTP 상태를 함께 기록하여 발음 표기 부재와 요청 실패를 구분합니다. 네트워크 예외나 성공 응답의 잘못된 구조는 그대로 오류로 전달합니다.

from urllib.parse import quote
import pandas as pd
from transforms.api import Input, Output, transform_pandas
from transforms.external.systems import ResolvedSource, Source, external_systems

DICTIONARY_SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
WORDS_INPUT = "/Examples/external/words"
PHONETICS_OUTPUT = "/Examples/external/enriched-phonetics"


@external_systems(dictionary_api_source=Source(DICTIONARY_SOURCE_RID))
@transform_pandas(Output(PHONETICS_OUTPUT), words_df=Input(WORDS_INPUT))
def enrich_words(dictionary_api_source: ResolvedSource, words_df: pd.DataFrame) -> pd.DataFrame:
    connection = dictionary_api_source.get_https_connection()
    client = connection.get_client()
    rows = []
    for word in words_df["word"]:
        response = client.get(
            connection.url + "/api/v2/entries/en/" + quote(word, safe=""),
            timeout=10,
        )
        if response.status_code == 200:
            phonetic = response.json()[0].get("phonetic")
            error = None
        else:
            phonetic = None
            error = f"HTTP {response.status_code}"
        rows.append({
            "word": word,
            "phonetic": phonetic,
            "http_status": response.status_code,
            "error": error,
        })
    return pd.DataFrame.from_records(
        rows, columns=["word", "phonetic", "http_status", "error"]
    ).astype({"word": "string", "phonetic": "string", "http_status": "Int64", "error": "string"})

입력이 비어 있어도 출력 열과 자료형을 정의합니다. HTTP 실패 행의 error를 확인하지 않은 채 모든 행이 정상적으로 보강되었다고 판단하지 않습니다.

단일 노드 실행

외부 연결 변환은 경량 단일 노드 실행과도 호환됩니다. 작은 데이터와 중간 규모 작업에서 Spark 실행 준비 비용을 줄일 수 있습니다. 다음 예제는 기존 @lightweight 방식으로 사전 API의 응답을 JSON 파일에 저장합니다.

from transforms.api import Output, lightweight, transform
from transforms.external.systems import Source, external_systems

DICTIONARY_SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
JSON_OUTPUT = "/Examples/external/dictionary-json"


@lightweight
@external_systems(dictionary_api_source=Source(DICTIONARY_SOURCE_RID))
@transform(output=Output(JSON_OUTPUT))
def fetch_dictionary_json(dictionary_api_source, output):
    connection = dictionary_api_source.get_https_connection()
    response = connection.get_client().get(
        connection.url + "/api/v2/entries/en/hello", timeout=10
    )
    response.raise_for_status()
    with output.filesystem().open("hello.json", "w") as destination:
        destination.write(response.text)

현재 단일 노드 API인 @transform.using(...).with_resources(...)와 소스 데코레이터를 결합하는 예제는 아래 Parquet 버퍼링 절에 있습니다.

사설망과 PostgreSQL sidecar 예제

사설망의 PostgreSQL에는 Foundry 워커 소스를 만들고, 네트워크 정책으로 Agent proxy와 사용할 에이전트를 선택합니다. 코드 가져오기를 허용하고 저장소에 소스를 추가합니다.

단순한 연결에 sidecar가 반드시 필요한 것은 아닙니다. 다음은 PostgreSQL 클라이언트를 컨테이너에 넣어 사용하는 예제입니다. Docker 이미지는 linux/amd64로 빌드하고 숫자형 비루트 사용자로 실행합니다. Artifacts 저장소의 유형을 Docker로 선택하여 이미지를 등록하고, 변환 저장소의 Settings > Libraries에 해당 저장소를 추가합니다. transforms-sidecar 라이브러리도 설치합니다.

컨테이너의 Dockerfile은 다음과 같습니다. Python 버전은 이 예제 컨테이너의 선택이며 Foundry 변환 런타임 버전과 별개입니다.

FROM python:3.11-slim
WORKDIR /usr/src/app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py ./
EXPOSE 1234
USER 5001
CMD ["python", "app.py"]

requirements.txt에는 다음 패키지를 넣습니다.

flask
psycopg2-binary

app.py는 실제 연결 값을 POST 본문으로 받아 기본 테이블 목록을 조회합니다. 이 서버는 sidecar 내부 호출용이며 외부 공개 API로 배포하는 구성이 아닙니다.

from contextlib import closing
from flask import Flask, jsonify, request
import psycopg2

app = Flask(__name__)


@app.route("/tables", methods=["POST"])
def list_tables():
    payload = request.get_json()
    parameters = {
        "dbname": payload["dbname"],
        "user": payload["user"],
        "password": payload["password"],
        "host": payload["host"],
        "port": payload["port"],
    }
    result = {"schema": [], "table": []}
    with closing(psycopg2.connect(**parameters)) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                "SELECT table_schema, table_name FROM information_schema.tables "
                "WHERE table_type = 'BASE TABLE'"
            )
            for schema_name, table_name in cursor.fetchall():
                result["schema"].append(schema_name)
                result["table"].append(table_name)
    return jsonify(result)


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=1234)

변환은 실제 소스에서 비밀번호를 읽고, 한 Spark 실행기에서 로컬 sidecar를 호출합니다. 클라이언트 객체나 소스 객체를 실행기에 전달하지 않고 연결 값 문자열만 전달합니다. 호스트와 데이터베이스, 사용자 및 비밀 정보 키는 실제 소스 구성에 맞춥니다. 이미지 이름 simple과 태그 0.0.1은 등록한 이미지와 일치해야 합니다.

import requests
from pyspark.sql import types as T
from transforms.api import Output, transform_df
from transforms.external.systems import Source, external_systems
from transforms.sidecar import sidecar

POSTGRES_SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000002"
TABLES_OUTPUT = "/Examples/external/postgres-tables"
POSTGRES_HOST = "postgres.example.com"
POSTGRES_DATABASE = "example_database"
POSTGRES_USER = "example_reader"
POSTGRES_PORT = 5432
PASSWORD_SECRET_KEY = "PASSWORD"


@external_systems(my_source=Source(POSTGRES_SOURCE_RID))
@sidecar(image="simple", tag="0.0.1")
@transform_df(Output(TABLES_OUTPUT))
def postgres_tables(my_source, ctx):
    parameters = {
        "host": POSTGRES_HOST,
        "port": POSTGRES_PORT,
        "dbname": POSTGRES_DATABASE,
        "user": POSTGRES_USER,
        "password": my_source.get_secret(PASSWORD_SECRET_KEY),
    }

    def query_sidecar(partition):
        for _ in partition:
            response = requests.post(
                "http://localhost:1234/tables", json=parameters, timeout=30
            )
            response.raise_for_status()
            yield (response.text,)

    rows = ctx.spark_session.sparkContext.parallelize([0], 1).mapPartitions(query_sidecar)
    schema = T.StructType([T.StructField("table_schema_json", T.StringType(), False)])
    return ctx.spark_session.createDataFrame(rows, schema)

메모리 규모를 추정하며 Parquet 쓰기

외부에서 읽은 모든 행을 목록에 모으면 메모리가 부족해질 수 있습니다. 다음 writer는 첫 행의 Python 객체 크기로 버퍼 크기를 추정하고, 임계값에 도달하면 Parquet 파일을 씁니다. 기본 임계값은 64 MiB입니다. 행 크기가 크게 달라지거나 Arrow 변환 중 추가 메모리가 필요할 수 있으므로 프로세스 메모리의 엄격한 상한을 보장하지는 않습니다.

예제는 실제 Dictionary API의 정의들을 읽어 word, part_of_speech, definition으로 씁니다. pyarrow와 pympler를 라이브러리에 추가합니다. 파일을 직접 기록한 뒤에는 put_metadata()로 스키마 추론을 마무리합니다. 빈 결과에도 스키마가 있는 빈 Parquet 파일을 기록합니다.

import gc
from datetime import datetime
import pyarrow as pa
import pyarrow.parquet as pq
from pympler import asizeof
from transforms.api import LightweightOutput, Output, transform
from transforms.external.systems import ResolvedSource, Source, external_systems

DICTIONARY_SOURCE_RID = "ri.magritte..source.00000000-0000-0000-0000-000000000001"
DEFINITIONS_OUTPUT = "/Examples/external/buffered-definitions"
OUTPUT_SCHEMA = pa.schema([
    ("word", pa.string()),
    ("part_of_speech", pa.string()),
    ("definition", pa.string()),
])


class BufferedParquetWriter:
    def __init__(self, output: LightweightOutput, schema: pa.Schema, flush_threshold_mb: int = 64):
        self.output = output
        self.schema = schema
        self.threshold = flush_threshold_mb * 1024 * 1024
        self.buffer = []
        self.estimated_bytes = 0
        self.row_size_estimate = None
        self.file_index = 0

    def append(self, row: dict):
        self.buffer.append(row)
        if self.row_size_estimate is None:
            self.row_size_estimate = asizeof.asizeof(row)
        self.estimated_bytes += self.row_size_estimate
        if self.estimated_bytes >= self.threshold:
            self.flush()

    def flush(self):
        if not self.buffer:
            return
        self.file_index += 1
        table = pa.Table.from_pylist(self.buffer, schema=self.schema)
        stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        with self.output.filesystem().open(f"output_{stamp}_{self.file_index}.parquet", "wb") as stream:
            pq.write_table(table, stream)
        self.buffer = []
        self.estimated_bytes = 0
        gc.collect()

    def finalize(self):
        self.flush()
        if self.file_index == 0:
            with self.output.filesystem().open("output_empty.parquet", "wb") as stream:
                pq.write_table(pa.Table.from_pylist([], schema=self.schema), stream)
        self.output.put_metadata()


@external_systems(my_source=Source(DICTIONARY_SOURCE_RID))
@transform.using(output=Output(DEFINITIONS_OUTPUT)).with_resources(memory_gb=4)
def buffered_definitions(my_source: ResolvedSource, output: LightweightOutput):
    output.set_mode("replace")
    writer = BufferedParquetWriter(output, OUTPUT_SCHEMA, flush_threshold_mb=64)
    connection = my_source.get_https_connection()
    client = connection.get_client()
    for word in ["apple", "dog", "cat"]:
        response = client.get(connection.url + "/api/v2/entries/en/" + word, timeout=10)
        response.raise_for_status()
        for entry in response.json():
            for meaning in entry["meanings"]:
                for definition in meaning["definitions"]:
                    writer.append({
                        "word": entry["word"],
                        "part_of_speech": meaning["partOfSpeech"],
                        "definition": definition["definition"],
                    })
    writer.finalize()

이 writer는 단일 프로세스에서 사용합니다. 여러 실행기가 같은 파일 이름을 동시에 쓰도록 확장하는 구성은 별도로 설계해야 합니다.

내보내기용 CSV 준비

CSV 파일을 원하는 외부 시스템에는 먼저 데이터셋을 CSV로 준비하고 Data Connection의 파일 내보내기를 연결할 수 있습니다. 준비 변환 자체는 외부 시스템에 연결하지 않으므로 소스 데코레이터가 필요하지 않습니다.

import csv
from transforms.api import Input, LightweightInput, LightweightOutput, Output, transform

SOURCE_DATA = "/Examples/external/prepared-data"
CSV_OUTPUT = "/Examples/external/csv-export"


@transform.using(source_data=Input(SOURCE_DATA), csv_output=Output(CSV_OUTPUT))
def prepare_csv(source_data: LightweightInput, csv_output: LightweightOutput):
    csv_output.set_mode("replace")
    frame = source_data.polars()
    with csv_output.filesystem().open("export.csv", "w") as destination:
        writer = csv.DictWriter(destination, fieldnames=frame.columns)
        writer.writeheader()
        for row in frame.iter_rows(named=True):
            writer.writerow(row)

이 예제는 입력을 Polars DataFrame으로 메모리에 읽습니다. 출력은 파일 내보내기에 사용하며, Parquet을 요구하는 테이블 내보내기와 구분합니다.

시스템별 구현 예제

같은 소스 연결 패턴을 다음 작업에 사용할 수 있습니다.

대상연결 및 처리 예제
REST APIClient credentials 토큰 교환, 자체 서명 인증서, 사용자 지정 HTTP 요청
Foundry APIFoundry 소스를 통한 플랫폼 API 호출
Amazon S3, SMB파일 읽기와 쓰기, 미디어 집합 수집
BigQuery, Databricks, Snowflake원천별 Python SDK와 세션 인증
SharePoint Online, SFTP, Kafka시스템별 SDK를 사용하는 수집과 쓰기
Directory(SSH)에이전트 호스트의 파일 수집
PostgreSQL, Microsoft SQL Server데이터베이스 클라이언트와 사용자 지정 쿼리

이전 방식과의 차이 및 이전 절차

소스 기반 방식에서는 연결 허용, 인증 정보, 외부 통신 정책과 반출 가능한 보안 표지를 각 소스에서 관리합니다. 저장소의 외부 시스템 탭은 자동으로 제공되며, 이전 방식처럼 저장소 설정을 ISO가 먼저 켜야 탭이 나타나는 구조가 아닙니다.

소스 설정이 바뀌면 코드 변경이나 버전 증가 없이 다음 빌드 시작 시 반영됩니다. 이미 실행 중인 빌드에는 소급 적용되지 않습니다. 여러 저장소가 같은 소스 구성을 공유할 수 있고, 인증 정보 회전과 사설망 연결, Data Lineage의 소스 연결 표시도 이 경로를 사용합니다.

수동 이전 절차는 다음과 같습니다.

  1. 기존 코드의 인증 정보, 외부 통신 정책과 반출 보안 표지를 확인합니다.
  2. 대응하는 Data Connection 소스를 만들거나 선택하고 코드 가져오기를 허용합니다.
  3. 기존 Python 저장소에 소스를 가져옵니다.
  4. @use_external_systems()를 @external_systems()와 소스 참조로 바꾸고, 인증 정보 접근도 소스의 메서드로 변경합니다.
  5. 브랜치에서 변경한 연결과 변환을 확인합니다.
  6. 병합한 뒤 더 이상 필요하지 않은 이전 저장소 외부 연결 설정을 끕니다.

자동 이전은 제공하지 않습니다. 같은 변환에 두 데코레이터를 함께 사용할 수 없습니다. 전체를 소스 기반으로 이전하는 것이 권장되며, 분리해야 하는 경우에는 각각 하나의 방식만 사용하는 변환으로 나눕니다.