Keyboard shortcuts

Press ← or → to navigate between chapters

Press ? to show this help

Press Esc to hide this help

Compute Module 만들고 실행하기

Foundry 폴더에서 + New > Compute Module을 선택하고 빈 Functions 모듈 또는 Pipeline 모듈을 만듭니다. 모듈의 Documentation 탭에는 해당 플랫폼 인스턴스에 맞는 저장소 생성 및 게시 안내가 있습니다.

Functions는 호출에 응답하고, Pipeline은 선언한 입력과 출력을 처리하며 스스로 실행합니다. 실행 모드와 권한을 먼저 정합니다.

로컬 Functions 모듈 준비

Docker와 사용할 언어 환경이 필요합니다. 아래 예제는 Python 3.12, Java 21 또는 Node.js 22를 사용합니다. Python SDK 0.34.0은 Python 3.10 이상, 4.0 미만을 요구합니다. Java Compute Module SDK는 Java 21 이상이며 TypeScript/Node.js SDK의 시작 환경은 Node.js 18 이상입니다.

새 폴더에 Dockerfile, 언어별 의존성 파일과 소스를 만듭니다. SDK를 쓰지 않는 언어라면 사용자 클라이언트를 구현합니다.

Python 예제

requirements.txt:

foundry-compute-modules==0.34.0

src/app.py:

from dataclasses import dataclass
from typing import TypedDict
from compute_modules.annotations import function

@dataclass
class AddPayload:
    x: int
    y: int

class HelloInput(TypedDict):
    name: str

@function
def add(context, event: AddPayload) -> str:
    return str(event.x + event.y)

@function
def hello(context, event: HelloInput) -> str:
    return "Hello " + event["name"]

Dockerfile:

FROM python:3.12
WORKDIR /app
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY src/app.py ./app.py
USER 5000
CMD ["python", "app.py"]

타입이 있는 입력과 반환값을 사용하므로 함수 명세를 추론할 수 있습니다. TypedDict는 사전으로, dataclass는 객체 속성으로 접근합니다. Python SDK에서 스트리밍, 인증과 로그를 설명합니다.

Java 예제

build.gradle:

plugins {
    id 'application'
    id 'com.gradleup.shadow' version '9.4.1'
}
repositories {
    mavenCentral()
}
dependencies {
    implementation 'com.palantir.computemodules:lib:0.6.0'
}
java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(21)
    }
}
application {
    mainClass = 'com.example.App'
}
tasks.named('shadowJar') {
    archiveFileName = 'compute-module.jar'
}

src/main/java/com/example/App.java:

package com.example;

import com.palantir.computemodules.ComputeModule;
import com.palantir.computemodules.functions.Context;

public final class App {
    public record AddEvent(int x, int y) {}
    public record HelloEvent(String name) {}

    public static void main(String[] args) {
        ComputeModule.builder()
                .add(App::add, AddEvent.class, Integer.class, "add")
                .add(App::hello, HelloEvent.class, String.class, "hello")
                .build()
                .start();
    }

    static Integer add(Context context, AddEvent event) {
        return event.x() + event.y();
    }

    static String hello(Context context, HelloEvent event) {
        return "Hello, " + event.name();
    }
}

Dockerfile:

FROM gradle:jdk21 AS build
WORKDIR /home/gradle/src
COPY build.gradle ./
COPY src ./src
RUN gradle shadowJar --no-daemon

FROM eclipse-temurin:21-jre-alpine
WORKDIR /app
COPY --from=build /home/gradle/src/build/libs/compute-module.jar ./app.jar
USER 5000
CMD ["java", "-jar", "/app/app.jar"]

application의 mainClass가 실행 JAR의 Main-Class로 들어갑니다. Shadow JAR 이름을 고정하여 런타임 이미지에는 그 파일만 복사합니다. Java SDK에서 입력 클래스와 로그 및 인증을 설명합니다.

JavaScript 예제

package.json:

{
  "name": "my-compute-module",
  "version": "0.0.1",
  "private": true,
  "dependencies": {
    "@palantir/compute-module": "0.2.14",
    "@sinclair/typebox": "^0.32.35"
  }
}

src/index.js:

const { ComputeModule } = require("@palantir/compute-module");
const { Type } = require("@sinclair/typebox");

const computeModule = new ComputeModule({
  definitions: {
    add: {
      input: Type.Object({ x: Type.Integer(), y: Type.Integer() }),
      output: Type.String(),
    },
    hello: {
      input: Type.Object({ name: Type.String() }),
      output: Type.String(),
    },
  },
});
computeModule.register("add", async ({ x, y }) => String(x + y));
computeModule.register("hello", async ({ name }) => "Hello " + name);
computeModule.default(async () => {
  throw new Error("Unsupported query name");
});

npm install로 package-lock.json을 생성하여 저장합니다. Dockerfile은 그 lockfile을 사용합니다.

FROM node:22-alpine
WORKDIR /app
COPY package.json package-lock.json ./
RUN npm ci --omit=dev
COPY src/index.js ./index.js
USER 5000
CMD ["node", "index.js"]

생성자가 수신을 시작하므로 start 메서드를 추가하지 않습니다. register 콜백의 인자는 data, context 순서입니다. TypeScript로 작성한다면 TypeScript 빌드 구성을 사용합니다.

언어별 입출력

Python과 Java 함수는 context, event 순서이며 TypeScript는 data, context 순서입니다. Python의 타입 없는 context는 딕셔너리이고 Java Context와 TypeScript QueryContext는 작업 ID를 제공합니다. 자격 증명은 실행 모드에 따라 별도로 조회합니다.

가상 입력 { "x": 1, "y": 2 }에 대해 위 Python/JavaScript add는 String을, Java add는 Integer를 반환하도록 정의했습니다. 다른 코드에서 가져올 때 선택한 구현의 출력 타입과 일치시킵니다. 반환값은 해당 SDK의 자료형 및 직렬화 조건에 맞아야 합니다.

이미지 게시와 Functions 실행

  1. 모듈의 Documentation > Build a compute module-backed function > Create your first container에서 Create or select repository를 선택합니다.
  2. Publish to DOCKER를 선택하고 표시한 레지스트리 로그인, 빌드 및 push 안내를 따릅니다. 이미지는 linux/amd64이며 latest 이외의 태그 또는 digest를 사용합니다.
  3. Configure > Add Container에서 저장소와 게시한 이미지를 선택합니다.
  4. Update configuration으로 저장합니다.
  5. Overview에서 모듈을 시작합니다.
  6. Responsive가 되면 아래 Query 패널로 함수를 호출하고 로그를 확인합니다.
  7. 플랫폼 앱에서 사용할 함수는 Functions 탭에서 Import합니다.

로컬 이미지 빌드는 다음과 같이 할 수 있습니다. 최종 게시 태그는 저장소의 Publish 안내에 맞춥니다.

docker build --platform linux/amd64 -t my-compute-module:0.0.1 .

실행되지 않으면 복제본 진단을 확인합니다.

Code Repositories에서 Python 작성

로컬 Docker 대신 Foundry의 Python Compute Module 저장소를 사용할 수 있습니다.

  1. 폴더에서 + New > Code Repository를 선택합니다.
  2. Python Compute Module 유형을 고르고 이름을 지정한 뒤 Create합니다.
  3. 생성된 src/app.py에서 함수를 정의합니다. 위 Python 예제의 TypedDict/dataclass 코드를 사용할 수 있습니다.
  4. Settings > Libraries에서 의존성을 추가하거나 저장소 루트의 meta.yaml에 의존성을 선언합니다.

  1. Tag version에서 예시 버전 0.0.1을 입력하고 Tag합니다. 이 버전으로 컨테이너 이미지를 생성합니다.

  1. 모듈의 Configure > Add Container에서 저장소와 생성한 태그 버전을 선택하고 Update configuration합니다.
  2. 모듈을 실행한 뒤 Functions > Import에서 발견한 함수를 선택해 등록합니다.

등록한 함수는 Workshop과 Slate 같은 앱에서 사용할 수 있습니다.

Pipeline의 스트림 처리 예제

가상 입력 스트림은 x라는 Integer 필드, 출력 스트림은 twice_x라는 Integer 필드를 가집니다. 모듈은 최근 레코드를 읽어 x를 두 배로 만들고 출력 스트림에 씁니다. 한 차례 처리한 뒤 60초 기다립니다.

이 예제는 최근 레코드 창을 반복 조회합니다. 소비자 offset이나 처리 완료 커서를 저장하지 않으므로 이미 처리한 레코드를 다시 쓸 수 있습니다. 단일 복제본을 전제로 하며 여러 복제본이 각자 같은 입력을 읽으면 중복 범위가 더 커집니다.

구성

  • Pipeline 모드를 선택합니다.
  • 같은 프로젝트의 입력/출력 스트리밍 데이터셋에 inputStream과 outputStream이라는 서로 다른 별칭을 붙입니다.
  • FOUNDRY_HOST에 실제 호스트를 설정하고 그 주소로 나가는 소스를 연결합니다.
  • 계속 실행해야 한다면 Minimum replicas를 1 이상으로 둡니다.
  • 두 배로 만든 값이 출력 Integer 스키마 범위에 들어오도록 입력 범위를 정합니다.

requirements.txt에는 requests를 넣습니다. Dockerfile은 위 Python 이미지 구성을 사용하고 src/app.py에는 다음 코드를 저장합니다.

import json
import os
from pathlib import Path
import time
from urllib.parse import quote
import requests


def process_record(record: dict) -> dict:
    return {"twice_x": record["value"]["x"] * 2}


def run_pipeline() -> None:
    with open(os.environ["RESOURCE_ALIAS_MAP"], encoding="utf-8") as file:
        aliases = json.load(file)
    input_info = aliases["inputStream"]
    output_info = aliases["outputStream"]
    host = os.environ["FOUNDRY_HOST"]

    def stream_url(resource: dict, endpoint: str) -> str:
        rid = quote(resource["rid"], safe="")
        branch = quote(resource.get("branch") or "master", safe="")
        return f"https://{host}/stream-proxy/api/streams/{rid}/branches/{branch}/{endpoint}"

    with requests.Session() as session:
        session.verify = os.environ["DEFAULT_CA_PATH"]
        while True:
            token = Path(os.environ["BUILD2_TOKEN"]).read_text().strip()
            headers = {"Authorization": f"Bearer {token}"}
            with session.get(
                stream_url(input_info, "records"), headers=headers, timeout=60
            ) as response:
                response.raise_for_status()
                records = response.json()
            for record in records:
                result = process_record(record)
                with session.post(
                    stream_url(output_info, "jsonRecord"),
                    json=result,
                    headers=headers,
                    timeout=60,
                ) as response:
                    response.raise_for_status()
            time.sleep(60)


if __name__ == "__main__":
    run_pipeline()

조회 응답의 각 레코드에 value.x가 있다는 입력 계약을 사용합니다. 필드 부재나 HTTP 실패가 발생하면 예외로 실행이 중단되며 실패한 전송을 성공한 것으로 처리하지 않습니다. 재시작 뒤에도 최근 창을 다시 읽으므로 이 코드를 정확히 한 번 처리하는 파이프라인으로 사용하지 않습니다.

이미지를 빌드하여 Artifacts에 게시하고 Pipeline 모듈의 컨테이너로 연결한 뒤 구성을 저장하고 Overview에서 시작합니다. Pipeline은 비대화형이므로 Query 패널로 시험하지 않습니다. 모듈 로그와 실제 출력 스트리밍 데이터셋에서 처리 상태를 확인합니다.