NodeCollection: 여러 시계열과 요약 묶기
foundryts.NodeCollection(*nodes, **kwargs)는 FunctionNode 또는 SummarizerNode의 컬렉션입니다. 반복 가능한 객체이므로 여러 시계열을 받는 함수에 전달할 수 있고, 각 노드에 같은 함수를 적용할 수도 있습니다.
원시 시계열이나 점 집합을 DataFrame으로 평가하면 series 열이 추가됩니다. 이 값은 동기화의 series ID 또는 points에서 지정한 이름입니다.
가상 시계열 두 개
다음 예제의 정수 시각은 나노초입니다. FoundryTS 환경에서 실행합니다.
from foundryts import Interval, NodeCollection
import foundryts.functions as F
series_1 = F.points(
(1, 100.0), (2, 200.0), (3, 300.0), name="series-1"
)
series_2 = F.points(
(1, 200.0), (2, 400.0), (3, 600.0), name="series-2"
)
collection = NodeCollection(series_1, series_2)
노드 목록을 전달하는 NodeCollection([series_1, series_2]) 형태도 사용할 수 있습니다. 가상 입력의 내용은 다음과 같습니다.
| series | 시각(ns) | value |
|---|---|---|
| series-1 | 1 | 100.0 |
| series-1 | 2 | 200.0 |
| series-1 | 3 | 300.0 |
| series-2 | 1 | 200.0 |
| series-2 | 2 | 400.0 |
| series-2 | 3 | 600.0 |
DataFrame의 timestamp는 1970-01-01 00:00:00 이후 해당 나노초를 나타내는 pandas.Timestamp로 표시됩니다.
함수의 입력 개수
단일 시계열을 받는 함수에 컬렉션을 직접 전달하면 오류가 발생합니다. 각 시계열의 값을 10배로 만들려면 map을 사용합니다.
scaled_collection = collection.map(F.scale(10))
scaled_df = scaled_collection.to_pandas()
결과에서 series-1의 값은 1000, 2000, 3000이고 series-2의 값은 2000, 4000, 6000입니다.
반면 scatter는 정확히 두 시계열을 입력으로 사용하므로 컬렉션을 그대로 전달합니다.
scatter_result = F.scatter()(collection)
scatter_df = scatter_result.to_pandas()
가상 예제의 결과는 다음 대응점을 포함하며 is_truncated는 False입니다.
| points.timestamp의 시각(ns) | points.first_value | points.second_value |
|---|---|---|
| 1 | 100.0 | 200.0 |
| 2 | 200.0 | 400.0 |
| 3 | 300.0 | 600.0 |
columns와 types
| 메서드 | 반환 형식 | 의미 |
|---|---|---|
| columns() | Tuple[str] | 결과 DataFrame의 열 이름 |
| types() | Tuple[Type] | 결과 열에 대응하는 형식 메타데이터 |
중첩 키는 점으로 연결합니다. 서로 다른 결과 형식을 가진 노드를 묶으면 각 요소의 열 합집합을 사용합니다.
raw_columns = collection.columns()
raw_types = collection.types()
combined_distribution = F.distribution()(collection)
distribution_columns = combined_distribution.columns()
두 원시 숫자형 시계열의 열은 series, timestamp, value입니다. 형식 메타데이터 예는 str, int, float이며 pandas의 dtype 객체와는 구분합니다. combined_distribution은 두 시계열 값의 합집합으로 분포 하나를 만듭니다.
분포와 통계처럼 다른 요약을 함께 묶을 수도 있습니다.
mixed = NodeCollection(
F.distribution()(series_1),
F.statistics()(series_2),
)
mixed_columns = mixed.columns()
mixed_types = mixed.types()
이 혼합 결과의 열은 다음 필드를 합쳐 구성합니다.
- 시계열 구분: series
- 분포: delta, distribution_values.count, distribution_values.start, distribution_values.end, start, end
- 공통 시간: start_timestamp, end_timestamp
- 통계: count, mean, earliest_point.timestamp, earliest_point.value, latest_point.timestamp, latest_point.value, smallest_point.timestamp, smallest_point.value, largest_point.timestamp, largest_point.value
map
NodeCollection.map(func)
각 시계열 노드에 func를 적용한 컬렉션을 반환합니다. func는 입력 노드를 받아 FunctionNode 또는 SummarizerNode를 만드는 FoundryTS 함수입니다. 사용할 함수가 해당 입력을 받을 수 있어야 합니다.
다음은 시계열별로 분포를 따로 계산합니다. 컬렉션 전체를 F.distribution에 전달하여 분포 하나를 만드는 경우와 구분합니다.
separate_distributions = collection.map(F.distribution())
separate_df = separate_distributions.to_pandas()
가상 입력에서 각 시계열 분포의 값은 다음과 같습니다. 표에는 관측이 있는 버킷을 표시합니다.
| series | delta | distribution_values.start | distribution_values.end | distribution_values.count |
|---|---|---|---|---|
| series-1 | 20.0 | 100.0 | 120.0 | 1 |
| series-1 | 20.0 | 200.0 | 220.0 | 1 |
| series-1 | 20.0 | 280.0 | 300.0 | 1 |
| series-2 | 40.0 | 200.0 | 240.0 | 1 |
| series-2 | 40.0 | 400.0 | 440.0 | 1 |
| series-2 | 40.0 | 560.0 | 600.0 | 1 |
각 결과에는 start와 end, start_timestamp와 end_timestamp도 포함됩니다. 이 예의 값 범위는 series-1이 100부터 300, series-2가 200부터 600입니다.
시간 구간별 분할
NodeCollection.map_intervals(intervals, interval_name=None)
| 인수 | 의미 |
|---|---|
| intervals | Interval 하나 또는 List[Interval] |
| interval_name | 결과의 구간 열에 사용할 선택적 이름 |
각 구간을 time_range로 적용하여 컬렉션의 모든 시계열을 해당 시간 범위로 나눕니다. 수동으로 만든 Interval이나 조건 검색 등에서 얻은 구간을 사용할 수 있습니다. 결과에는 구간 시작과 끝을 나타내는 열이 추가됩니다.
intervals = [
Interval(1, 2),
Interval(2, 3),
Interval(3, 4),
Interval(1, 3),
]
windowed = collection.map_intervals(intervals, interval_name="interval")
scaled_windowed = windowed.map(F.scale(1000))
windowed_df = scaled_windowed.to_pandas()
결과에는 interval, interval.start와 interval.end 열이 포함됩니다. 이 예제에는 별도 구간 이름을 지정하지 않았습니다. 시작은 포함하고 끝은 제외하므로 겹치는 구간에 속한 관측은 구간마다 나타납니다.
| series | interval.start | interval.end | 시각(ns) | 원래 값 | 1000배 값 |
|---|---|---|---|---|---|
| series-1 | 1 | 2 | 1 | 100.0 | 100000.0 |
| series-1 | 2 | 3 | 2 | 200.0 | 200000.0 |
| series-1 | 3 | 4 | 3 | 300.0 | 300000.0 |
| series-1 | 1 | 3 | 1 | 100.0 | 100000.0 |
| series-1 | 1 | 3 | 2 | 200.0 | 200000.0 |
| series-2 | 1 | 2 | 1 | 200.0 | 200000.0 |
| series-2 | 2 | 3 | 2 | 400.0 | 400000.0 |
| series-2 | 3 | 4 | 3 | 600.0 | 600000.0 |
| series-2 | 1 | 3 | 1 | 200.0 | 200000.0 |
| series-2 | 1 | 3 | 2 | 400.0 | 400000.0 |
pandas로 평가
NodeCollection.to_pandas(parallel=4, mode='thread')
컬렉션의 모든 질의를 평가하고 pandas.DataFrame 하나로 결합합니다. parallel은 병렬 스레드 또는 프로세스 수이며 기본값은 4입니다. 1이면 다중 처리를 하지 않습니다. mode는 thread 또는 process를 선택하며 기본값은 thread입니다.
각 질의 결과를 로컬에 저장한 뒤 결합하므로 전체 결과가 메모리에 들어가야 합니다. 병렬 수를 늘리면 메모리와 자원 사용량도 증가합니다. process 모드는 추가 오버헤드가 있으며 과도한 병렬화는 성능을 낮출 수 있습니다.
result_df = collection.to_pandas(parallel=1)
Spark로 평가
NodeCollection.to_dataframe(numPartitions=16)
모든 시계열과 계산 결과를 평가하여 pyspark.sql.DataFrame으로 결합합니다. 많은 원시 관측, FunctionNode 결과나 여러 요약 결과를 분산 처리할 때 사용합니다.
numPartitions는 시계열 데이터를 Spark 실행기에 나눌 파티션 수이며 기본값은 16입니다. 실행기가 많으면 큰 값이 병렬 처리에 도움이 될 수 있고, 실행기가 적으면 작은 값이 효율적일 수 있습니다. 컬렉션 크기, 실행기 수와 메모리에 맞춰 정합니다.
spark_df = collection.to_dataframe(numPartitions=16)
원시 시계열 결과의 series 열에는 시계열의 series ID나 points에서 지정한 이름이 들어갑니다. pandas와 Spark 중 후속 처리에 필요한 방식을 선택합니다.
이전 RDD 출력
NodeCollection.to_rdd(numPartitions=16)
버전 1.0.0부터 사용 중단이 진행 중인 메서드입니다. 컬렉션을 평가하여 (key, object) 쌍의 pyspark.RDD를 반환합니다. 그룹화 또는 윈도 연산의 결과이면 key에 메타데이터 키나 구간 정보가 들어가고, 그렇지 않으면 노드의 시계열 식별자를 사용합니다. numPartitions는 Spark 실행 파티션 수이며 기본값은 16입니다.