연산자
연산자 (Operators)
연산자(operators)는 하나 이상의 DataStream을 변환해서 새로운 DataStream을 만듭니다. 프로그램은 여러 변환을 결합해 정교한 데이터플로우 토폴로지를 구성할 수 있습니다.
출처: 문서
본문
이 섹션은 기본 변환, 그 변환들을 적용한 후의 효율적인 물리 파티셔닝, 그리고 Flink의 연산자 체이닝(operator chaining)에 대한 통찰을 제공합니다.
DataStream 변환 (DataStream Transformations)
Map
DataStream → DataStream
하나의 요소를 받아 하나의 요소를 만듭니다. 입력 스트림의 값을 두 배로 만드는 map 함수:
DataStream<Integer> dataStream = //...
dataStream.map(new MapFunction<Integer, Integer>() {
@Override
public Integer map(Integer value) throws Exception {
return 2 * value;
}
});
data_stream = env.from_collection(collection=[1, 2, 3, 4, 5])
data_stream.map(lambda x: 2 * x, output_type=Types.INT())
FlatMap
DataStream → DataStream
하나의 요소를 받아 0개, 1개 또는 그 이상의 요소를 만듭니다. 문장을 단어로 나누는 flatmap 함수:
dataStream.flatMap(new FlatMapFunction<String, String>() {
@Override
public void flatMap(String value, Collector<String> out)
throws Exception {
for(String word: value.split(" ")){
out.collect(word);
}
}
});
data_stream = env.from_collection(collection=['hello apache flink', 'streaming compute'])
data_stream.flat_map(lambda x: x.split(' '), output_type=Types.STRING())
Filter
DataStream → DataStream
각 요소에 대해 boolean 함수를 평가하고 함수가 true를 반환하는 요소만 유지합니다. 0 값을 걸러내는 filter:
dataStream.filter(new FilterFunction<Integer>() {
@Override
public boolean filter(Integer value) throws Exception {
return value != 0;
}
});
data_stream = env.from_collection(collection=[0, 1, 2, 3, 4, 5])
data_stream.filter(lambda x: x != 0)
KeyBy
DataStream → KeyedStream
스트림을 서로 분리된 파티션으로 논리적으로 분할합니다. 같은 키를 가진 모든 레코드는 같은 파티션에 할당됩니다. 내부적으로 *keyBy()*는 해시 파티셔닝으로 구현됩니다. 키 지정에는 여러 방법이 있습니다.
dataStream.keyBy(value -> value.getSomeKey());
dataStream.keyBy(value -> value.f0);
data_stream = env.from_collection(collection=[(1, 'a'), (2, 'a'), (3, 'b')])
data_stream.key_by(lambda x: x[1], key_type=Types.STRING()) // Key by the result of KeySelector
다음의 경우 타입은 키가 될 수 없습니다:
- POJO 타입이지만
hashCode()메서드를 재정의하지 않고Object.hashCode()구현에 의존하는 경우. - 어떤 타입의 배열인 경우.
Reduce
KeyedStream → DataStream
키드 데이터 스트림에 대한 "롤링" reduce입니다. 현재 요소를 마지막 reduce 값과 결합해 새 값을 방출합니다. 부분 합계의 스트림을 만드는 reduce 함수:
keyedStream.reduce(new ReduceFunction<Integer>() {
@Override
public Integer reduce(Integer value1, Integer value2)
throws Exception {
return value1 + value2;
}
});
data_stream = env.from_collection(collection=[(1, 'a'), (2, 'a'), (3, 'a'), (4, 'b')], type_info=Types.TUPLE([Types.INT(), Types.STRING()]))
data_stream.key_by(lambda x: x[1]).reduce(lambda a, b: (a[0] + b[0], b[1]))
Window
KeyedStream → WindowedStream
윈도우는 이미 파티셔닝된 KeyedStream에 정의할 수 있습니다. 윈도우는 각 키의 데이터를 어떤 특성(예: 지난 5초 안에 도착한 데이터)에 따라 그룹화합니다. 윈도우에 대한 완전한 설명은 windows를 참고하세요.
dataStream
.keyBy(value -> value.f0)
.window(TumblingEventTimeWindows.of(Duration.ofSeconds(5)));
data_stream.key_by(lambda x: x[1]).window(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
WindowAll
DataStream → AllWindowedStream
윈도우는 일반 DataStream에도 정의할 수 있습니다. 윈도우는 모든 스트림 이벤트를 어떤 특성에 따라 그룹화합니다. 윈도우에 대한 완전한 설명은 windows를 참고하세요.
이것은 많은 경우 비병렬 변환입니다. windowAll 연산자를 위해 모든 레코드가 하나의 작업(task)에 모입니다.
dataStream
.windowAll(TumblingEventTimeWindows.of(Duration.ofSeconds(5)));
data_stream.window_all(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
Window Apply
WindowedStream → DataStream / AllWindowedStream → DataStream
윈도우 전체에 일반 함수를 적용합니다. 아래는 윈도우의 요소를 수동으로 합산하는 함수입니다. windowAll 변환을 사용한다면 대신 AllWindowFunction을 사용해야 합니다.
windowedStream.apply(new WindowFunction<Tuple2<String,Integer>, Integer, Tuple, Window>() {
public void apply (Tuple tuple,
Window window,
Iterable<Tuple2<String, Integer>> values,
Collector<Integer> out) throws Exception {
int sum = 0;
for (value t: values) {
sum += t.f1;
}
out.collect (new Integer(sum));
}
});
// non-keyed window stream에 AllWindowFunction 적용
allWindowedStream.apply (new AllWindowFunction<Tuple2<String,Integer>, Integer, Window>() {
public void apply (Window window,
Iterable<Tuple2<String, Integer>> values,
Collector<Integer> out) throws Exception {
int sum = 0;
for (value t: values) {
sum += t.f1;
}
out.collect (new Integer(sum));
}
});
class MyWindowFunction(WindowFunction[tuple, int, int, TimeWindow]):
def apply(self, key: int, window: TimeWindow, inputs: Iterable[tuple]) -> Iterable[int]:
sum = 0
for input in inputs:
sum += input[1]
yield sum
class MyAllWindowFunction(AllWindowFunction[tuple, int, TimeWindow]):
def apply(self, window: TimeWindow, inputs: Iterable[tuple]) -> Iterable[int]:
sum = 0
for input in inputs:
sum += input[1]
yield sum
windowed_stream.apply(MyWindowFunction())
# non-keyed window stream에 AllWindowFunction 적용
all_windowed_stream.apply(MyAllWindowFunction())
WindowReduce
WindowedStream → DataStream
윈도우에 함수형 reduce 함수를 적용하고 reduce된 값을 반환합니다.
windowedStream.reduce (new ReduceFunction<Tuple2<String,Integer>>() {
public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
return new Tuple2<String,Integer>(value1.f0, value1.f1 + value2.f1);
}
});
class MyReduceFunction(ReduceFunction):
def reduce(self, value1, value2):
return value1[0], value1[1] + value2[1]
windowed_stream.reduce(MyReduceFunction())
Union
DataStream* → DataStream
둘 이상의 데이터 스트림을 합쳐 모든 스트림의 모든 요소를 포함하는 새 스트림을 만듭니다. 참고: 데이터 스트림을 자기 자신과 union하면 결과 스트림에서 각 요소를 두 번 얻습니다.
dataStream.union(otherStream1, otherStream2, ...);
data_stream.union(otherStream1, otherStream2, ...)
Window Join
DataStream,DataStream → DataStream
주어진 키와 공통 윈도우에서 두 데이터 스트림을 join합니다.
dataStream.join(otherStream)
.where(<key selector>).equalTo(<key selector>)
.window(TumblingEventTimeWindows.of(Duration.ofSeconds(3)))
.apply (new JoinFunction () {...});
# 이 기능은 Python에서 아직 지원되지 않습니다.
Interval Join
KeyedStream,KeyedStream → DataStream
공통 키를 가진 두 키드 스트림의 두 요소 e1, e2를 주어진 시간 간격에 대해 join해서 e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound가 되게 합니다.
// 이렇게 하면 두 스트림을 join합니다:
// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2
keyedStream.intervalJoin(otherKeyedStream)
.between(Duration.ofMillis(-2), Duration.ofMillis(2)) // lower and upper bound
.upperBoundExclusive(true) // optional
.lowerBoundExclusive(true) // optional
.process(new IntervalJoinFunction() {...});
# 이 기능은 Python에서 아직 지원되지 않습니다.
Window CoGroup
DataStream,DataStream → DataStream
주어진 키와 공통 윈도우에서 두 데이터 스트림을 cogroup합니다.
dataStream.coGroup(otherStream)
.where(0).equalTo(1)
.window(TumblingEventTimeWindows.of(Duration.ofSeconds(3)))
.apply (new CoGroupFunction () {...});
# 이 기능은 Python에서 아직 지원되지 않습니다.
Connect
DataStream,DataStream → ConnectedStream
타입을 유지하면서 두 데이터 스트림을 "연결(connects)"합니다. Connect는 두 스트림 간에 공유 상태를 허용합니다.
DataStream<Integer> someStream = //...
DataStream<String> otherStream = //...
ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream);
stream_1 = ...
stream_2 = ...
connected_streams = stream_1.connect(stream_2)
CoMap, CoFlatMap
ConnectedStream → DataStream
연결된 데이터 스트림의 map 및 flatMap과 유사합니다.
connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() {
@Override
public Boolean map1(Integer value) {
return true;
}
@Override
public Boolean map2(String value) {
return false;
}
});
connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() {
@Override
public void flatMap1(Integer value, Collector<String> out) {
out.collect(value.toString());
}
@Override
public void flatMap2(String value, Collector<String> out) {
for (String word: value.split(" ")) {
out.collect(word);
}
}
});
class MyCoMapFunction(CoMapFunction):
def map1(self, value):
return value[0] + 1, value[1]
def map2(self, value):
return value[0], value[1] + 'flink'
class MyCoFlatMapFunction(CoFlatMapFunction):
def flat_map1(self, value)
for i in range(value[0]):
yield i
def flat_map2(self, value):
yield value[0] + 1
connectedStreams.map(MyCoMapFunction())
connectedStreams.flat_map(MyCoFlatMapFunction())
Cache
DataStream → CachedDataStream
변환의 중간 결과를 캐시합니다. 현재 배치 실행 모드로 실행되는 작업만 지원됩니다. 캐시된 중간 결과는 중간 결과가 처음 계산될 때 지연 생성되어 이후 작업에서 재사용될 수 있습니다. 캐시가 손실되면 원래 변환을 사용해 다시 계산됩니다.
DataStream<Integer> dataStream = //...
CachedDataStream<Integer> cachedDataStream = dataStream.cache();
cachedDataStream.print(); // cachedDataStream으로 원하는 작업 수행
...
env.execute(); // 실행하고 캐시 생성
cachedDataStream.print(); // 캐시된 결과 소비
env.execute();
data_stream = ... # DataStream
cached_data_stream = data_stream.cache()
cached_data_stream.print()
# ...
env.execute() # 실행하고 캐시 생성
cached_data_stream.print() # 캐시된 결과 소비
env.execute()
Full Window Partition
DataStream → PartitionWindowedStream
각 파티션의 모든 레코드를 별도로 전체 윈도우로 수집하고 처리합니다. 윈도우 방출은 입력의 끝에서 트리거됩니다. 이 접근 방식은 주로 배치 처리 시나리오에 적합합니다. non-keyed DataStream의 경우 파티션은 서브태스크의 모든 레코드를 포함합니다. KeyedStream의 경우 파티션은 키의 모든 레코드를 포함합니다.
DataStream<Integer> dataStream = //...
PartitionWindowedStream<Integer> partitionWindowedDataStream = dataStream.fullWindowPartition();
// PartitionWindowedStream으로 full window partition 처리
DataStream<Integer> resultStream = partitionWindowedDataStream.mapPartition(
new MapPartitionFunction<Integer, Integer>() {
@Override
public void mapPartition(
Iterable<Integer> values, Collector<Integer> out) {
int result = 0;
for (Integer value : values) {
result += value;
}
out.collect(result);
}
}
);
물리 파티셔닝 (Physical Partitioning)
Flink는 (원한다면) 변환 후 정확한 스트림 파티셔닝에 대한 저수준 제어도 다음 함수들을 통해 제공합니다.
Custom Partitioning
DataStream → DataStream
사용자 정의 Partitioner를 사용해 각 요소에 대한 대상 작업을 선택합니다.
dataStream.partitionCustom(partitioner, "someKey");
dataStream.partitionCustom(partitioner, 0);
data_stream = env.from_collection(collection=[(2, 'a'), (2, 'a'), (3, 'b')])
data_stream.partition_custom(lambda key, num_partition: key % partition, lambda x: x[0])
Random Partitioning
DataStream → DataStream
요소를 균일 분포에 따라 무작위로 파티셔닝합니다.
dataStream.shuffle();
data_stream.shuffle()
Rescaling
DataStream → DataStream
요소를 다운스트림 연산의 하위 집합에 라운드-로빈으로 파티셔닝합니다. 예를 들어 소스의 각 병렬 인스턴스에서 여러 mapper의 하위 집합으로 fan out해 부하를 분산하지만 rebalance()가 초래하는 전체 재균형은 원하지 않는 파이프라인을 원할 때 유용합니다. 이를 위해서는 다른 구성 값(예: TaskManager의 슬롯 수)에 따라 네트워크를 통한 데이터 전송 대신 로컬 데이터 전송만 필요할 수 있습니다.
업스트림 연산이 요소를 보내는 다운스트림 연산의 하위 집합은 업스트림과 다운스트림 연산의 병렬도에 따라 달라집니다. 예를 들어 업스트림 연산의 병렬도가 2이고 다운스트림 연산의 병렬도가 6이라면, 한 업스트림 연산은 3개의 다운스트림 연산에 요소를 분배하고 다른 업스트림 연산은 나머지 3개에 분배합니다. 반대로 다운스트림 연산의 병렬도가 2이고 업스트림의 병렬도가 6이라면, 3개의 업스트림 연산이 한 다운스트림 연산에 분배하고 나머지 3개가 다른 다운스트림 연산에 분배합니다.
서로 다른 병렬도가 서로의 배수가 아닌 경우 하나 이상의 다운스트림 연산은 업스트림 연산에서 서로 다른 수의 입력을 받게 됩니다.
위 예제의 연결 패턴 시각화는 그림을 참고하세요.
dataStream.rescale();
data_stream.rescale()
Broadcasting
DataStream → DataStream
요소를 모든 파티션에 브로드캐스트합니다.
dataStream.broadcast();
data_stream.broadcast()
작업 체이닝과 리소스 그룹 (Task Chaining and Resource Groups)
두 개의 연속 변환을 체이닝하면 성능 향상을 위해 같은 스레드 내에 함께 배치하는 것을 의미합니다. Flink는 가능하면 기본적으로 연산자를 체이닝합니다(예: 두 개의 연속 map 변환). API는 원한다면 체이닝에 대한 세밀한 제어를 제공합니다:
전체 작업에서 체이닝을 비활성화하려면 StreamExecutionEnvironment.disableOperatorChaining()을 사용하세요. 더 세밀한 제어를 위해 다음 함수들을 사용할 수 있습니다. 이 함수들은 이전 변환을 가리키므로 DataStream 변환 직후에만 사용할 수 있습니다. 예를 들어 someStream.map(...).startNewChain()은 사용할 수 있지만 someStream.startNewChain()은 사용할 수 없습니다.
리소스 그룹은 Flink의 슬롯입니다(슬롯 참고). 원한다면 연산자를 별도 슬롯에 수동으로 격리할 수 있습니다.
Start New Chain
이 연산자부터 시작해 새 체인을 시작합니다. 두 mapper는 체이닝되고, filter는 첫 번째 mapper에 체이닝되지 않습니다.
someStream.filter(...).map(...).startNewChain().map(...);
some_stream.filter(...).map(...).start_new_chain().map(...)
Disable Chaining
map 연산자를 체이닝하지 않습니다.
someStream.map(...).disableChaining();
some_stream.map(...).disable_chaining()
Set Slot Sharing Group
연산의 슬롯 공유 그룹을 설정합니다. Flink는 같은 슬롯 공유 그룹을 가진 연산을 같은 슬롯에 배치하고, 슬롯 공유 그룹이 없는 연산은 다른 슬롯에 유지합니다. 이는 슬롯을 격리하는 데 사용할 수 있습니다. 모든 입력 연산이 같은 슬롯 공유 그룹에 있다면 슬롯 공유 그룹은 입력 연산에서 상속됩니다. 기본 슬롯 공유 그룹의 이름은 "default"이며, slotSharingGroup("default")를 호출해 연산을 명시적으로 이 그룹에 넣을 수 있습니다.
someStream.filter(...).slotSharingGroup("name");
some_stream.filter(...).slot_sharing_group("name")
이름과 설명 (Name And Description)
Flink의 연산자와 작업 vertex는 이름(name)과 설명(description)을 가집니다. 이름과 설명 모두 연산자 또는 작업 vertex가 무엇을 하는지에 대한 소개지만 용도가 다릅니다.
연산자와 작업 vertex의 이름은 웹 UI, 스레드 이름, 로깅, 메트릭 등에 사용됩니다. 작업 vertex의 이름은 그 안의 연산자 이름을 기반으로 구성됩니다. 이름은 외부 시스템에 대한 높은 부하를 피하기 위해 가능한 한 간결해야 합니다.
설명은 실행 계획에 사용되며 웹 UI에서 작업 vertex의 세부 정보로 표시됩니다. 작업 vertex의 설명은 그 안의 연산자 설명을 기반으로 구성됩니다. 설명은 런타임 디버깅을 돕기 위해 연산자에 대한 세부 정보를 포함할 수 있습니다.
someStream.filter(...).name("filter").setDescription("x in (1, 2, 3, 4) and y > 1");
some_stream.filter(...).name("filter").set_description("x in (1, 2, 3, 4) and y > 1")
작업 vertex의 설명 형식은 기본적으로 트리 형식 문자열입니다. 이전 버전에서처럼 설명을 계단식 형식으로 설정하려면 pipeline.vertex-description-mode를 CASCADING으로 설정할 수 있습니다.
Flink SQL이 생성하는 연산자는 기본적으로 연산자 타입과 id로 구성된 이름과 상세한 설명을 가집니다. 이전 버전에서처럼 이름을 상세 설명으로 설정하려면 table.exec.simplify-operator-name-enabled를 false로 설정할 수 있습니다.
파이프라인 토폴로지가 복잡할 때 pipeline.vertex-name-include-index-prefix를 true로 설정해 vertex 이름에 토폴로지 인덱스를 추가할 수 있습니다. 그러면 로그나 메트릭 태그로 그래프에서 vertex를 쉽게 찾을 수 있습니다.