상태 변경, replay와 증분 실행 강제
행 수준 변환은 한 행만으로 결과를 만들 수 있습니다. 곱셈이나 Filter가 해당합니다. 상태를 사용하는 함수는 여러 행의 정보를 유지하며 결과를 계산합니다. 대표적인 상태 연산은 다음과 같습니다.
| 연산 | 지원 범위 |
|---|---|
| Aggregate | 배치 증분 집계이며 스트리밍에서는 Aggregate over Window를 사용합니다. |
| Outer caching join | 스트리밍에서 사용합니다. |
| Heartbeat detection | 스트리밍에서 사용합니다. |
| Time bounded drop duplicates | 스트리밍에서 사용합니다. |
| Time bounded event time sort | 스트리밍에서 사용합니다. |
예를 들어 짝수만 골라 합산하다가 홀수만 고르도록 바꾸면 기존의 짝수 합과 이후의 홀수 합이 한 상태에 섞입니다. 새 로직에 맞는 합계를 얻으려면 입력을 다시 처리하여 상태를 재구성해야 합니다. 이런 변경을 state break라고 합니다.
Replay 범위
Replay from start of input data는 스트림의 시작 또는 증분 데이터셋의 첫 트랜잭션부터 다시 처리합니다.
Replay from amount of time ago는 스트리밍에서만 사용할 수 있습니다. 지정한 과거 시각 직전에 commit된 첫 트랜잭션부터 그 이후의 데이터를 포함합니다. 따라서 지정 시각보다 앞선 데이터가 트랜잭션 하나 분량만큼 들어올 수 있습니다.
일반적인 replay는 선택 사항이지만 상태 호환성이 깨지면 시스템이 이를 감지하고 배포 시 replay를 요구할 수 있습니다.
문제는 Errors 탭에 나타나며 해결 전에는 배포가 차단됩니다. Advanced settings는 기본적으로 접혀 있고 replay 옵션도 선택 해제되어 있습니다. 전략을 선택하면 Confirm replay strategy에서 동작과 하위 영향을 확인해야 합니다.
기본 replay는 출력 스트림의 이력을 다시 구성하며 하위 소비 파이프라인에도 replay가 필요할 수 있습니다. 재처리 구간이 크면 수일의 중단이 생길 수 있습니다. 배포 이벤트의 timeline에는 감지한 상태 변경과 선택한 replay 전략이 작업 행으로 기록됩니다.
변경별 영향
| 변경 | 상태 처리 |
|---|---|
| 상태 연산 수정 또는 제거 | 전체 replay가 필요합니다. 해당 연산을 공급하는 입력의 제거도 상태 일관성에 영향을 줍니다. |
| 입력 추가 | 새 입력은 처음부터 읽고 기존 입력은 마지막 처리 위치에서 계속할 수 있습니다. 변경 인지가 필요하며, 뒤의 상태 연산과 과거 데이터가 논리적으로 맞는지 판단해야 합니다. |
| 입력 또는 출력 제거 | 그 입력이나 출력의 상태를 처리 클러스터에서 제거하고 replay 없이 계속할 수 있습니다. 다만 job group 내부 변경이나 상태 연산의 일관성 문제는 예외입니다. |
| 출력 추가 | State break가 발생하며 입력 시작 또는 지정 과거 시각에서 replay해야 합니다. |
| 출력 열 추가 | Replay가 필요하지 않습니다. |
| 출력 열 제거 | 이전 데이터와 새 출력 스키마가 달라져 replay가 필요합니다. |
| 상위 파이프라인 replay | 입력이 다시 구성되므로 현재 파이프라인의 상태도 재처리해야 할 수 있습니다. |
입력 또는 출력 변경을 replay 없이 수용할 수 있을 때는 state-break 창에서 변경을 인지하고 계속할 수 있습니다. 이 선택은 입력 변경이 뒤의 상태 연산과 논리적으로 맞는지 자동으로 보장하지 않습니다.
Job group 안의 입력이나 출력을 제거하면 이 인지 옵션을 사용할 수 없으며 시작점 또는 지정 과거 시각에서 replay해야 합니다. 단순 입력 교체라도 상태 집계가 이전 원천에 의존했다면 일관성을 위해 replay가 필요할 수 있습니다.
입력 스키마는 배포할 때 고정됩니다. 이후 원천 스키마가 바뀌어도 직접 재배포하기 전까지 이전 스키마로 읽습니다.
상위 출력이 초기화된 경우
상위 스트리밍 파이프라인이 기본값인 Reset Outputs on replay를 사용하면 출력은 초기화되고 선택한 replay 구간의 데이터로 다시 채워집니다. 하위 파이프라인은 기간을 따로 맞추는 대신 Start of input data를 선택해야 합니다. 새 상위 출력의 처음부터 읽어 객체 같은 하위 결과를 전체 재처리 구간에 맞게 다시 구성합니다.
기존 출력을 보존하며 재수집
출력 view를 초기화하지 않고 특정 시점부터 입력을 다시 읽을 수도 있습니다. 배포 패널의 Advanced에서 replay할 때 Reset Outputs on replay를 끕니다.
이때 재수집을 시작할 당시 출력에 있던 데이터는 유지됩니다. 입력을 다시 읽는 범위와 출력에 이미 있는 데이터를 어떻게 합칠지는 해당 변환 및 출력의 쓰기 방식에 맞게 고려해야 합니다.
Require incremental execution
증분 실행을 의도한 job이 스냅샷으로 바뀌는 것을 막으려면 Require incremental execution을 켭니다. 증분 조건을 충족하지 못하면 자동으로 전체 새로 고침을 수행하는 대신 job이 실패합니다.
입력의 우발적인 스냅샷, 출력 스키마 변경으로 인한 스냅샷 등 예상하지 않은 전체 처리를 방지할 때 사용합니다. PySpark의 @incremental(require_incremental=True)에 대응하는 기능입니다.
Deploy옆 또는 오른쪽 도구 모음의Build settings를 엽니다.Advanced configuration으로 이동합니다.Require incremental execution을True로 설정합니다. 기본값은 비활성 상태인No value입니다.
설정은 job group과 관계없이 파이프라인의 모든 증분 출력에 적용됩니다. 증분 입력이나 출력이 하나 이상 있어야 하며, 없는 파이프라인에서 켜면 배포 오류가 발생합니다.
예외적으로 파이프라인 변경 자체에 state break가 필요한 경우에는 변경을 인지한 뒤 스냅샷으로 실행할 수 있습니다. 따라서 이 설정을 켰다는 이유로 명시적으로 승인한 상태 재구성까지 불가능해지는 것은 아닙니다.
빈 트랜잭션 commit
증분 빌드가 출력 행을 하나도 만들지 않을 수 있습니다. Commit empty transactions를 켜면 이 경우에도 빈 트랜잭션을 commit하여 해당 입력을 처리했다는 상태를 남깁니다. 끈 상태에서는 결과를 만들지 못한 입력을 이후 빌드에서 다시 처리할 수 있습니다.
Build settings > Advanced configuration에서 Commit empty transactions를 활성화합니다. 정상적인 필터 결과가 빈 출력일 수 있는 증분 파이프라인에서 유용합니다.