브로드캐스트 상태 패턴
브로드캐스트 상태 패턴 (The Broadcast State Pattern)
이 섹션에서는 실제로 브로드캐스트 상태(broadcast state)를 사용하는 방법을 배워요. 상태 기반 스트림 처리의 개념은 Stateful Stream Processing을 참고해요.
출처: 문서
본문
제공 API (Provided APIs)
제공되는 API를 보여주기 위해 전체 기능을 소개하기 전에 먼저 예제로 시작할게요. 예제로 서로 다른 색과 모양의 객체 스트림이 있고, 어떤 패턴을 따르는 같은 색 객체의 쌍을 찾고 싶은 경우를 사용할게요. 예: 사각형 뒤에 삼각형. 흥미로운 패턴의 집합은 시간이 지나면서 진화한다고 가정해요.
이 예제에서 첫 번째 스트림은 Color와 Shape 프로퍼티를 가진 Item 타입 요소를 포함하고, 다른 스트림은 Rules를 포함해요.
Items 스트림에서 시작해 같은 색의 쌍을 원하므로 그냥 Color로 *키잉(key it)*하면 돼요. 이렇게 하면 같은 색의 요소가 같은 물리 머신에 있게 됩니다.
Java:
// key the items by color
KeyedStream<Item, Color> colorPartitionedStream = itemStream
.keyBy(new KeySelector<Item, Color>(){...});
Python:
# key the items by color
color_partitioned_stream = item_stream.key_by(lambda item: ...)
Rules로 넘어가면, Rules를 담은 스트림은 모든 다운스트림 작업에 브로드캐스트되어야 하며, 이 작업들은 Rules를 로컬에 저장해 들어오는 모든 Item에 대해 평가할 수 있어야 해요. 아래 snippet은 i) 규칙 스트림을 브로드캐스트하고 ii) 제공된 MapStateDescriptor로 Rules가 저장될 브로드캐스트 상태를 만듭니다.
Java:
// a map descriptor to store the name of the rule (string) and the rule itself.
MapStateDescriptor<String, Rule> ruleStateDescriptor = new MapStateDescriptor<>(
"RulesBroadcastState",
BasicTypeInfo.STRING_TYPE_INFO,
TypeInformation.of(new TypeHint<Rule>() {}));
// broadcast the rules and create the broadcast state
BroadcastStream<Rule> ruleBroadcastStream = ruleStream
.broadcast(ruleStateDescriptor);
Python:
# a map descriptor to store the name of the rule (string) and the rule (Python object) itself.
rule_state_descriptor = MapStateDescriptor("RuleBroadcastState", Types.STRING(), Types.PICKLED_BYTE_ARRAY())
# broadcast the rules and create the broadcast state
rule_broadcast_stream = rule_stream.broadcast(rule_state_descriptor)
마지막으로 Item 스트림의 들어오는 요소에 대해 Rules를 평가하려면:
- 두 스트림을 연결하고,
- 일치 감지 로직을 지정해야 해요.
스트림(키드 또는 비키드)을 BroadcastStream과 연결하는 것은 BroadcastStream을 인자로 하여 비브로드캐스트 스트림에서 connect()를 호출해 수행할 수 있어요. 이는 BroadcastConnectedStream을 반환하며, 여기서 특수한 종류의 CoProcessFunction으로 process()를 호출할 수 있어요. 함수는 우리의 일치 로직을 담을 것입니다. 함수의 정확한 타입은 비브로드캐스트 스트림의 타입에 따라 달라져요:
- keyed이면 함수는 KeyedBroadcastProcessFunction
- non-keyed이면 함수는 BroadcastProcessFunction
비브로드캐스트 스트림이 keyed이므로 다음 snippet에는 위 호출이 포함돼요. connect는 BroadcastStream을 인자로 하여 비브로드캐스트 스트림에서 호출해야 해요.
Java:
DataStream<String> output = colorPartitionedStream
.connect(ruleBroadcastStream)
.process(
// type arguments in our KeyedBroadcastProcessFunction represent:
// 1. the key of the keyed stream
// 2. the type of elements in the non-broadcast side
// 3. the type of elements in the broadcast side
// 4. the type of the result, here a string
new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
// my matching logic
}
);
Python:
class MyKeyedBroadcastProcessFunction(KeyedBroadcastProcessFunction):
# my matching logic
...
output = color_partitioned_stream \
.connect(rule_broadcast_stream) \
.process(MyKeyedBroadcastProcessFunction())
BroadcastProcessFunction과 KeyedBroadcastProcessFunction
CoProcessFunction의 경우처럼 이 함수들에는 구현할 두 process 메서드가 있어요: 브로드캐스트된 스트림의 들어오는 요소를 처리하는 processBroadcastElement()와 비브로드캐스트 요소를 처리하는 processElement(). 메서드의 전체 시그니처는 아래에 제시돼요.
Java:
public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {
public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
}
public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {
public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
}
Python:
class BroadcastProcessFunction(BaseBroadcastProcessFunction, Generic[IN1, IN2, OUT]):
@abstractmethod
def process_element(value: IN1, ctx: ReadOnlyContext):
pass
@abstractmethod
def process_broadcast_element(value: IN2, ctx: Context):
pass
class KeyedBroadcastProcessFunction(BaseBrodcastProcessFunction, Generic[KEY, IN1, IN2, OUT]):
@abstractmethod
def process_element(value: IN1, ctx: ReadOnlyContext):
pass
@abstractmethod
def process_broadcast_element(value: IN2, ctx: Context):
pass
def on_timer(timestamp: int, ctx: OnTimerContext):
pass
가장 먼저 주목할 것은 두 함수 모두 브로드캐스트 쪽 요소 처리를 위한 processBroadcastElement() 메서드와 비브로드캐스트 쪽 요소를 위한 processElement() 구현을 요구한다는 것이에요. 두 메서드는 제공되는 컨텍스트가 다릅니다. 비브로드캐스트 쪽은 ReadOnlyContext를, 브로드캐스트 쪽은 Context를 가져요.
이 두 컨텍스트(아래 열거의 ctx)는:
Java:
- 브로드캐스트 상태에 접근:
ctx.getBroadcastState(MapStateDescriptor<K, V> stateDescriptor) - 요소의 타임스탬프 조회 허용:
ctx.timestamp() - 현재 워터마크 가져오기:
ctx.currentWatermark() - 현재 처리 시간 가져오기:
ctx.currentProcessingTime() - 사이드 출력으로 요소 발행:
ctx.output(OutputTag<X> outputTag, X value)
Python:
- 브로드캐스트 상태에 접근:
ctx.get_broadcast_state(state_descriptor: MapStateDescriptor) - 요소의 타임스탬프 조회 허용:
ctx.timestamp() - 현재 워터마크 가져오기:
ctx.current_watermark() - 현재 처리 시간 가져오기:
ctx.current_processing_time() - 사이드 출력으로 요소 발행:
yield output_tag, value
getBroadcastState()의 stateDescriptor는 위의 .broadcast(ruleStateDescriptor)의 것과 동일해야 해요.
차이는 각각이 브로드캐스트 상태에 주는 접근 유형에 있어요. 브로드캐스트 쪽은 읽기-쓰기 접근을, 비브로드캐스트 쪽은 읽기 전용 접근(그래서 이름이 그렇습니다)을 가져요. 이유는 Flink에는 작업 간 통신이 없기 때문이에요. 따라서 Broadcast State의 내용이 우리 연산자의 모든 병렬 인스턴스에서 동일함을 보장하기 위해 모든 작업이 같은 요소를 보는 브로드캐스트 쪽에만 읽기-쓰기 접근을 주고, 그 쪽의 각 들어오는 요소에 대한 계산이 모든 작업에서 동일하도록 요구해요. 이 규칙을 무시하면 상태의 일관성 보장이 깨져, 일관되지 않고 종종 디버깅하기 어려운 결과가 발생해요.
processBroadcastElement()에 구현된 로직은 모든 병렬 인스턴스에서 동일한 결정적 동작을 가져야 해요!
마지막으로 KeyedBroadcastProcessFunction이 keyed 스트림에서 동작하기 때문에 BroadcastProcessFunction에서 사용할 수 없는 일부 기능을 노출해요. 즉:
processElement()의 ReadOnlyContext는 Flink의 기본 타이머 서비스에 접근을 제공하며, 이를 통해 이벤트 및/또는 처리 시간 타이머를 등록할 수 있어요. 타이머가 발화하면onTimer()가 ReadOnlyContext와 동일한 기능에 더해 다음을 노출하는 OnTimerContext로 호출돼요:- 발화된 타이머가 이벤트 시간인지 처리 시간인지 물을 수 있는 능력
- 타이머와 연관된 키 조회
processBroadcastElement()의 Context는applyToKeyedState(StateDescriptor<S, VS> stateDescriptor, KeyedStateFunction<KS, S> function)메서드를 포함해요. 이는 제공된 stateDescriptor와 연관된 모든 키의 모든 상태에 적용될 KeyedStateFunction을 등록할 수 있게 해요.apply_to_keyed_state는 아직 PyFlink에서 지원되지 않는다는 점을 주의해요.
타이머 등록은 KeyedBroadcastProcessFunction의 processElement()에서만 가능하며 그 곳에서만 가능해요. 브로드캐스트 요소와 연관된 키가 없으므로 processBroadcastElement() 메서드에서는 불가능해요.
원래 예제로 돌아가면, 우리의 KeyedBroadcastProcessFunction은 다음과 같을 수 있어요.
Java:
new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
// store partial matches, i.e. first elements of the pair waiting for their second element
// we keep a list as we may have many first elements waiting
private final MapStateDescriptor<String, List<Item>> mapStateDesc =
new MapStateDescriptor<>(
"items",
BasicTypeInfo.STRING_TYPE_INFO,
new ListTypeInfo<>(Item.class));
// identical to our ruleStateDescriptor above
private final MapStateDescriptor<String, Rule> ruleStateDescriptor =
new MapStateDescriptor<>(
"RulesBroadcastState",
BasicTypeInfo.STRING_TYPE_INFO,
TypeInformation.of(new TypeHint<Rule>() {}));
@Override
public void processBroadcastElement(Rule value,
Context ctx,
Collector<String> out) throws Exception {
ctx.getBroadcastState(ruleStateDescriptor).put(value.name, value);
}
@Override
public void processElement(Item value,
ReadOnlyContext ctx,
Collector<String> out) throws Exception {
final MapState<String, List<Item>> state = getRuntimeContext().getMapState(mapStateDesc);
final Shape shape = value.getShape();
for (Map.Entry<String, Rule> entry :
ctx.getBroadcastState(ruleStateDescriptor).immutableEntries()) {
final String ruleName = entry.getKey();
final Rule rule = entry.getValue();
List<Item> stored = state.get(ruleName);
if (stored == null) {
stored = new ArrayList<>();
}
if (shape == rule.second && !stored.isEmpty()) {
for (Item i : stored) {
out.collect("MATCH: " + i + " - " + value);
}
stored.clear();
}
// there is no else{} to cover if rule.first == rule.second
if (shape.equals(rule.first)) {
stored.add(value);
}
if (stored.isEmpty()) {
state.remove(ruleName);
} else {
state.put(ruleName, stored);
}
}
}
}
Python:
class MyKeyedBroadcastProcessFunction(KeyedBroadcastProcessFunction):
def __init__(self):
self._map_state_desc = MapStateDescriptor("item", Types.STRING(), Types.LIST(Types.PICKLED_BYTE_ARRAY()))
self._rule_state_desc = MapStateDescriptor("RulesBroadcastState", Types.STRING(), Types.PICKLED_BYTE_ARRAY())
self._map_state = None
def open(self, ctx: RuntimeContext):
self._map_state = ctx.get_map_state(self._map_state_desc)
def process_broadcast_element(value: Rule, ctx: KeyedBroadcastProcessFunction.Context):
ctx.get_broadcast_state(self._rule_state_desc).put(value.name, value)
def process_element(value: Item, ctx: KeyedBroadcastProcessFunction.ReadOnlyContext):
shape = value.get_shape()
for rule_name, rule in ctx.get_broadcast_state(self._rule_state_desc).items():
stored = self._map_state.get(rule_name)
if stored is None:
stored = []
if shape == rule.second and len(stored) > 0:
for i in stored:
yield "MATCH: {} - {}".format(i, value)
stored = []
if shape == rule.first:
stored.append(value)
if len(stored) == 0:
self._map_state.remove(rule_name)
else:
self._map_state.put(rule_name, stored)
중요 고려사항 (Important Considerations)
제공 API를 설명한 후, 이 섹션은 브로드캐스트 상태를 사용할 때 명심해야 할 중요한 것들에 초점을 맞춰요:
- 작업 간 통신이 없음: 앞서 말했듯이 이것이 (Keyed)-BroadcastProcessFunction의 브로드캐스트 쪽만 브로드캐스트 상태의 내용을 수정할 수 있는 이유예요. 또한 사용자는 모든 작업이 각 들어오는 요소에 대해 브로드캐스트 상태의 내용을 같은 방식으로 수정하도록 해야 해요. 그렇지 않으면 작업마다 내용이 달라 일관되지 않은 결과가 발생할 수 있어요.
- 브로드캐스트 상태에서 이벤트 순서가 작업마다 다를 수 있음: 스트림의 요소를 브로드캐스트하는 것은 모든 요소가 (결국) 모든 다운스트림 작업으로 간다는 것을 보장하지만, 요소는 각 작업에 다른 순서로 도착할 수 있어요. 따라서 각 들어오는 요소의 상태 갱신은 들어오는 이벤트의 순서에 의존해서는 안 됩니다.
- 모든 작업이 브로드캐스트 상태를 체크포인트함: 체크포인트 시 모든 작업이 브로드캐스트 상태에 같은 요소를 갖지만(체크포인트 배리어는 요소를 추월하지 않음), 하나만이 아니라 모든 작업이 브로드캐스트 상태를 체크포인트해요. 이는 복원 중 모든 작업이 같은 파일을 읽는 것(핫스팟 방지)을 피하기 위한 설계 결정이지만, 체크포인트된 상태 크기를 p(= 병렬도) 배만큼 증가시키는 비용이 들어요. Flink는 복원/재확장 시 중복 없음과 누락 데이터 없음을 보장해요. 같거나 더 작은 병렬도로 복구할 때 각 작업은 자신의 체크포인트된 상태를 읽어요. 확장 시 각 작업은 자신의 상태를 읽고 나머지 작업(p_new-p_old)은 이전 작업의 체크포인트를 라운드로빈 방식으로 읽어요.
- RocksDB 상태 백엔드 없음: 브로드캐스트 상태는 런타임에 메모리에 유지되며 그에 따라 메모리 프로비저닝을 해야 해요. 이는 모든 연산자 상태에 해당해요.