시계열 파이프라인의 고급 구성

일반적인 구성은 Pipeline Builder를 사용하는 것이 권장됩니다. 시계열에 필요한 변환 최적화를 자동으로 적용합니다. 저수준 제어나 아직 제공되지 않는 기능 때문에 Code Repositories를 사용하려면 먼저 Palantir 담당자와 고급 구성을 확인합니다.

수동 구성은 입력 최적화, 동기화 생성, 객체 기반 데이터셋 생성과 TSP 연결로 이루어집니다.

데이터셋 입력 최적화

입력에는 series ID, 측정 시각과 값이 필요합니다. 아래는 가상 측정 데이터이며 timestamp는 Timestamp형입니다.

실시간으로 새 데이터가 들어오는 배치 데이터셋은 보통 증분 빌드로 관리하여 반복 계산과 수집 후 조회까지의 지연을 줄입니다. 일반적인 동기화에서는 같은 시계열의 값을 같은 데이터셋에 둡니다.

Python 정렬 및 저장

다음 입력과 출력은 가상 경로입니다. 입력 열은 위 표와 같은 series_id, timestamp, value로 준비합니다.

from transforms.api import Input, Output, transform


@transform(
    output_dataset=Output("/Examples/TimeSeries/optimized"),
    input_dataset=Input("/Examples/TimeSeries/raw"),
)
def write_time_series(output_dataset, input_dataset):
    dataframe = (
        input_dataset.dataframe()
        .repartitionByRange("series_id")
        .sortWithinPartitions("series_id", "timestamp")
    )
    output_dataset.write_dataframe(dataframe, output_format="soho")

Java 정렬 및 저장

같은 입력 계약의 Java 변환입니다.

package myproject.datasets;

import com.palantir.foundry.spark.api.DatasetFormatSettings;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

public final class TimeSeriesWriter {
    @Compute
    public void writePartitioned(
            @Input("/Examples/TimeSeries/raw") FoundryInput inputDataset,
            @Output("/Examples/TimeSeries/optimized") FoundryOutput outputDataset) {
        Dataset<Row> input = inputDataset.asDataFrame().read();
        Dataset<Row> output = input
                .repartitionByRange(input.col("series_id"))
                .sortWithinPartitions("series_id", "timestamp");

        outputDataset.getDataFrameWriter(output)
                .setFormatSettings(DatasetFormatSettings.builder()
                        .format("soho")
                        .build())
                .write();
    }
}

Range 재파티션만으로 각 파티션 안의 행이 정렬되지는 않으므로 sortWithinPartitions도 적용합니다. 아직 프로젝션이 만들어지지 않은 새 데이터를 시계열에서 읽을 수 있도록 수동 데이터셋은 Soho 형식으로 저장합니다.

예제처럼 파티션 수를 생략하면 Spark의 설정값을 사용합니다. 실제 파이프라인에서는 실행기 메모리와 필요한 병렬성에 맞춰 수를 정합니다.

• 가능한 적은 파티션을 사용합니다.
• 파티션 크기는 128MB보다 크게 구성하는 것이 권장됩니다.
• 일반적으로 파티션당 행 수는 50억 미만으로 둡니다.

파티션은 실행기에 들어갈 만큼 작아야 하고 목표 지연을 달성할 만큼 병렬로 처리할 수도 있어야 합니다. 더 많은 작은 파티션은 빌드 시간을 줄일 수 있지만 시계열 조회에 유리한 큰 파티션 구성과 절충합니다.

View 최적화

많은 데이터를 시간, 일 또는 주 단위 데이터셋으로 나누고 하나의 View로 묶은 뒤 동기화 하나에 연결할 수 있습니다. View가 전체 구간을 포함하므로 이 구성에서는 같은 series ID가 여러 기반 데이터셋에 나뉘어 있어도 됩니다.

1. 분할한 기반 데이터셋을 포함하는 View를 만듭니다.
2. View를 동기화 입력으로 지정합니다.
3. Advanced settings에서 Index view dataset inputs를 켭니다.

기반 데이터셋 수에 따른 조회 비용과 권장 규모는 동기화 설정을 확인합니다. 이 경로는 같은 ID를 여러 독립 동기화에 나누는 구성과 다릅니다.

스트리밍 입력 최적화

스트림 키는 Series ID 하나만 사용합니다. 다른 열을 키에 추가하지 않아야 시계열 색인에 맞는 순서를 유지하고 읽을 파티션을 줄일 수 있습니다. 높은 처리량이 필요하면 파티션을 최소 세 개로 구성하는 것이 권장됩니다.

Streaming pipeline

시계열 동기화를 출력으로 두고 Key by 변환에 Series ID만 지정합니다. Build settings > Advanced configuration > Number of Output Partitions에서 처리량에 맞는 출력 파티션 수를 정합니다.

Streaming sync

Streaming sync를 사용하는 경로에서는 Configuration > Keys에 Series ID를 지정합니다. Throughput에서 Very high를 선택하고 필요한 파티션 수를 설정합니다.

이미 레코드를 수집한 스트림의 키 또는 파티션 구성을 바꾸려면 스트림 재설정이 필요합니다. 기존 수집 데이터와 하위 사용 영향을 검토한 뒤 변경 절차를 수행합니다.

동기화와 객체 연결

설정 도우미에서 Go to Builder를 선택하거나 Time Series Catalog에서 동기화를 구성합니다.

Ontology Manager로 돌아와 동기화를 TSP의 기반 데이터 원천으로 추가합니다. 같은 동기화에 필요한 ID가 모두 있으면 여러 TSP가 그 동기화를 사용할 수 있습니다. Save로 온톨로지 변경을 저장합니다.

새 객체 유형의 첫 저장이라면 초기 색인이 끝나야 TSP를 분석할 수 있습니다. 객체 유형의 Datasources 탭에서 색인 상태를 확인합니다.

객체 기반 데이터셋과 유형

객체 기반 데이터 스키마에 맞춰 객체 하나당 한 행을 준비합니다. 측정 데이터와 같은 변환에서 series ID의 고유 집합을 추출하고 객체 메타데이터를 연결할 수도 있습니다. 증분 파이프라인에서는 새 ID를 기존 메타데이터와 병합하고 추가하는 구성을 사용합니다.

준비한 데이터셋에서 객체 유형을 만들거나 Dataset Preview의 All actions > Create object type을 사용합니다. 생성 후 TSP로 사용할 속성과 동기화를 지정합니다.