Advanced Transformations

상위: Flink

요약

Flink의 고급 변환 연산자들을 다룹니다. union, connect와 같은 멀티 스트림 결합부터 rebalance, rescale과 같은 병렬 작업 최적화 함수까지, 복잡한 스트림 처리 시나리오를 위한 고급 연산자들의 사용법을 학습합니다.

멀티 스트림 및 결합 함수

union

stream1 = env.from_collection([1, 2, 3], type_info=Types.INT())
stream2 = env.from_collection([4, 5, 6], type_info=Types.INT())

union_stream = stream1.union(stream2)

result_union = list(union_stream.execute_and_collect())
print("Union 결과:", result_union)

🟢 결과: [1, 2, 3, 4, 5, 6]

connect

stream_str = env.from_collection(["A", "B"], type_info=Types.STRING())
stream_int = env.from_collection([1, 2], type_info=Types.INT())

connected_stream = stream_str.connect(stream_int)

class MyCoMapFunction(CoMapFunction):
    def map1(self, value):
        return f"String: {value}"
    def map2(self, value):
        return f"Int: {value}"

co_mapped_stream = connected_stream.map(MyCoMapFunction(), output_type=Types.STRING())

result_connect = list(co_mapped_stream.execute_and_collect())
print("Connect 결과:", result_connect)

🟢 결과: ['String: A', 'String: B', 'Int: 1', 'Int: 2']

병렬 작업 최적화 함수

rebalance()

env.set_parallelism(2)  # 병렬 작업 두 개 설정

# 예제 입력: 0부터 9까지 숫자
data = list(range(10))

# Source 병렬성 1로 설정 (한 개의 병렬 인스턴스가 모든 데이터를 가짐)
stream = env.from_collection(data, type_info=Types.INT()).set_parallelism(1)

# rebalance() 적용: 데이터를 라운드로빈으로 재분배
rebalanced = stream.rebalance()

# 각 병렬 인스턴스에서 받은 데이터 확인
rebalanced.map(lambda x: f"[rebalance] 데이터: {x}").print()

env.execute("rebalance 예제")

🟢 결과:

1> [rebalance] 데이터: 1
2> [rebalance] 데이터: 0
1> [rebalance] 데이터: 3
2> [rebalance] 데이터: 2
1> [rebalance] 데이터: 5
2> [rebalance] 데이터: 4
1> [rebalance] 데이터: 7
2> [rebalance] 데이터: 6
1> [rebalance] 데이터: 9
2> [rebalance] 데이터: 8

rescale()

env.set_parallelism(2)

data = list(range(10))  # 0부터 9까지 숫자

stream = env.from_collection(data, type_info=Types.INT()).set_parallelism(1)

# rescale() 적용: 동일 호스트 내 우선적으로 분배
rescaled = stream.rescale()

rescaled.map(lambda x: f"[rescale] 데이터: {x}").print()

env.execute("rescale 예제")

🟢 결과:

2> [rescale] 데이터: 1
1> [rescale] 데이터: 0
2> [rescale] 데이터: 3
1> [rescale] 데이터: 2
2> [rescale] 데이터: 5
1> [rescale] 데이터: 4
2> [rescale] 데이터: 7
1> [rescale] 데이터: 6
2> [rescale] 데이터: 9
1> [rescale] 데이터: 8

Rebalance vs Rescale

disable_chaining()

env.set_parallelism(1)

data = [1, 2, 3, 4, 5]
stream = env.from_collection(data, type_info=Types.INT())

# 첫 번째 map만 체이닝 해제
result = stream.map(lambda x: x + 1).disable_chaining() \
               .map(lambda x: x * 2) \
               .map(lambda x: f"[Unchained] 결과: {x}")

result.print()

# 실행 계획 정보 확인
plan = env.get_execution_plan()
print("\n[체이닝 해제된 실행 계획]\n", plan)

env.execute("Unchained Map Example")

🟢 결과:

{
  "id": 2,
  "type": "Map",
  "pact": "Operator",
  "contents": "Map",
  "parallelism": 1,
  "predecessors": [
    {
      "id": 1,
      "ship_strategy": "FORWARD",
      "side": "second"
    }
  ]
},
{
  "id": 8,
  "type": "Map, Map, Map",
  "pact": "Operator",
  "contents": "Map, Map, Map",
  "parallelism": 1,
  "predecessors": [
    {
      "id": 2,
      "ship_strategy": "FORWARD",
      "side": "second"
    }
  ]
}

• 첫 번째 map 연산은 단독으로 실행됨 ("contents": "Map")
• 이후 map 연산 둘은 체이닝되어 함께 실행됨 ("contents": "Map, Map")

Java에 있지만 PyFlink에는 없는 함수

관련 개념

• Transformations - 기본 변환 연산자들
• Stream Iteration - 스트림 분할 및 반복 처리
• DataStream API - DataStream API 기본 구조