센서 스트림을 만들고 Pipeline Builder에서 변환하기
이 예제는 온도 레코드를 스트림에 보내고, 센서 ID를 대문자로 정리한 뒤 온도 값이 3을 초과하는 행만 새 스트림으로 내보냅니다. 입력 스키마는 sensor_id: String, temperature: Double이며 두 열은 null을 허용하지 않는 구성입니다.
입력 스트림 생성
프로젝트에서 + New > Stream을 선택합니다.
Define 화면에서 처리량은 Normal, 파티션은 하나인 예제 구성을 사용하고 위의 두 열을 정의합니다. Create stream을 누르면 연결 화면으로 이동합니다.
이 예제는 상태 없는 행 변환이므로 키별 순서에 의존하지 않습니다. 상태 집계나 CDC를 추가하려면 파티션 수 하나만으로 순서를 보장한다고 가정하지 말고 키 설정을 적용합니다.
시험 레코드 발행
Connect via API에서 Curl (Bash)를 선택합니다. 짧은 시험에는 Test with a personal token으로 단기 개인 토큰을 만들고 안내에 따라 입력한 뒤 Next Step으로 진행할 수 있습니다. 운영 수집에는 개인 토큰 대신 OAuth 토큰 흐름을 사용합니다.
아래 Bash 예제는 공개 V2 발행 API를 사용합니다. 실제 Foundry HTTPS 기본 URL, 새 스트림의 데이터셋 RID와 브랜치를 각각 FOUNDRY_URL, DATASET_RID, STREAM_BRANCH 환경 변수에 설정합니다. Bash, jq와 --fail-with-body 및 --header @-를 지원하는 cURL이 필요합니다. 토큰은 화면에 표시하지 않는 입력으로 받고 인증 헤더는 표준 입력으로 전달합니다.
(
set -euo pipefail
set +x
: "${FOUNDRY_URL:?Foundry HTTPS 기본 URL을 설정하세요}"
: "${DATASET_RID:?대상 스트림 데이터셋 RID를 설정하세요}"
: "${STREAM_BRANCH:?대상 스트림 브랜치 이름을 설정하세요}"
BASE_URL="${FOUNDRY_URL%/}"
RID_PATH="$(printf '%s' "$DATASET_RID" | jq -sRr @uri)"
BRANCH_PATH="$(printf '%s' "$STREAM_BRANCH" | jq -sRr @uri)"
trap 'unset ACCESS_TOKEN' EXIT
IFS= read -r -s -p '시험용 액세스 토큰: ' ACCESS_TOKEN
printf '\n'
printf 'Authorization: Bearer %s\n' "$ACCESS_TOKEN" |
curl --silent --show-error --fail-with-body \
--request POST \
--header @- \
--header 'Content-Type: application/json' \
--data-binary '{"records":[{"sensor_id":" sensor-01 ","temperature":1.234},{"sensor_id":" sensor-02 ","temperature":4.132}]}' \
"$BASE_URL/api/v2/highScale/streams/datasets/$RID_PATH/streams/$BRANCH_PATH/publishRecords"
)
두 레코드는 가상 시험 입력입니다. V2 본문은 records 목록 안에 열 값을 직접 넣습니다. 다른 발행 API가 사용하는 value 래퍼 배열과 섞지 않습니다. 발행할 권한과 OAuth의 api:streams-write 범위, 운영용 토큰 교환은 푸시 기반 수집에서 설명합니다.
발행 후 입력 스트림의 레코드 화면에서 두 필드와 값을 확인합니다. 이 API는 처리된 출력 레코드를 응답으로 돌려주는 기능이 아닙니다.
스트리밍 파이프라인 생성
입력 스트림의 Start pipelining을 선택하고 새 파이프라인 대화상자에서 Streaming pipeline, Create Pipeline을 선택합니다.
입력 노드가 그래프에 추가됩니다. 노드를 선택하면 미리보기를 볼 수 있지만 이 미리보기는 스트림의 콜드 저장소 뷰를 사용하므로 새 레코드가 즉시 나타나지 않을 수 있습니다.
세 변환 추가
입력 노드의 Transform 또는 파란 T 아이콘을 선택합니다. 입력 열의 자료형에 따라 지원되는 스트리밍 변환이 표시됩니다.
Uppercase에서sensor_id를 선택하고Apply를 누릅니다.Trim whitespace에서도sensor_id를 선택하고 적용합니다.Filter > Keep rows에서temperature를 고르고greater than값3을 지정합니다.
Apply all changes를 선택한 뒤 Back to graph로 돌아옵니다.
출력과 배포
변환 경로 노드를 선택하고 New dataset으로 출력 스트림을 만듭니다.
먼저 Save를 선택한 뒤 Deploy > Deploy pipeline을 실행합니다. 배포 전에는 구성과 필요한 리소스의 준비 상태를 검증합니다. 실패하면 대화상자의 각 메시지를 확인하여 해결하고 다시 저장 및 배포합니다.
저장만 하고 배포하지 않으면 실행 중인 변환 로직은 새 변경으로 갱신되지 않습니다.
출력 노드를 선택하고 아래 Data preview 위에 표시된 스트림 이름을 열어 출력 스트림 화면으로 이동합니다.
클러스터가 시작하는 데 약 1분이 걸릴 수 있습니다. 실행 상태가 된 뒤 같은 시험 발행을 다시 수행하여 새 레코드를 보내면 변환 경로를 확인할 수 있습니다. 계산 규칙에 따르면 1.234인 행은 제외되고 4.132인 행은 남으며, 남은 센서 ID는 SENSOR-02가 됩니다.