9. Data Ordering: Bin Pack Orderer & FIFO Orderer#

1
정렬 전:                    정렬 후:
2
Key=1 Time=10:00           Key=1 Time=09:04
3
Key=2 Time=09:04           Key=1 Time=10:00
4
Key=1 Time=09:04           Key=1 Time=10:04
5
Key=1 Time=10:04           Key=2 Time=09:04
6
Key=3 Time=10:15           Key=2 Time=10:00
7
Key=2 Time=10:00           Key=3 Time=10:15

1
Bin 1: [Key=1 09:04, Key=2 09:04, Key=3 10:15]   ← 각 키의 첫 번째
2
Bin 2: [Key=1 10:00, Key=2 10:00]                  ← 각 키의 두 번째
3
Bin 3: [Key=1 10:04]                               ← 각 키의 세 번째

1
from pyspark.sql import functions as F
2

3
# Step 1: 파티션 내 로컬 정렬 (shuffle 없음)
4
sorted_events = events.sortWithinPartitions(
5
    [F.col("visit_id"), F.col("event_time")]
6
)
7

8
# Step 2+3: 파티션별로 bin 생성 후 순차 전송
9
sorted_events.foreachPartition(
10
    lambda rows: write_records_to_kinesis("output-stream", rows)
11
)

1
# Bin 생성 로직
2
def write_records_to_kinesis(output_stream, visits_rows):
3
    producer = boto3.client("kinesis")
4
    delivery_bins = []
5
    bin_index = 0
6
    last_visit_id = None
7

8
    for visit in visits_rows:
9
        if visit.visit_id != last_visit_id:
10
            # 새 키 → bin index 리셋
11
            last_visit_id = visit.visit_id
12
            bin_index = 0
13

14
        # bin이 부족하면 새로 생성
15
        if len(delivery_bins) <= bin_index:
16
            delivery_bins.append([])
17

18
        delivery_bins[bin_index].append(visit)
19
        bin_index += 1
20

21
    # 순차 전송: bin 1 완료 → bin 2 → bin 3
22
    for bin_records in delivery_bins:
23
        send_bulk_with_retry(producer, output_stream, bin_records)

1
# Kafka: produce 후 즉시 flush → ack 대기
2
producer.produce(topic, key=key, value=value)
3
producer.flush()  # 브로커 응답까지 블로킹

1
# Kafka: 버퍼링 허용하되 동시 요청을 1개로 제한
2
producer = Producer({
3
    "max.in.flight.requests.per.connection": 1,
4
    "queue.buffering.max.ms": 1000    # 1초간 버퍼링
5
})
6
producer.produce(topic, key=key, value=value)
7
# flush 없음 — 프로듀서가 자동으로 bulk 전송

1
# Kafka: 멱등 프로듀서 — 동시 요청 5개까지 순서 보장
2
producer = Producer({
3
    "max.in.flight.requests.per.connection": 5,
4
    "enable.idempotence": True,
5
    "queue.buffering.max.ms": 2000
6
})
7
producer.produce(topic, key=key, value=value)

1
previous_seq = None
2
for record in records_to_deliver:
3
    result = client.put_record(
4
        StreamName="output-stream",
5
        Data=record.data,
6
        PartitionKey=record.key,
7
        SequenceNumberForOrdering=previous_seq  # 이전 시퀀스 참조
8
    )
9
    previous_seq = result["SequenceNumber"]

1
for message in messages:
2
    producer.send(message)    # 성공
3
    consumer.ack(message)     # ← 여기서 실패하면?
4
# 재시작 시 이미 전송된 message를 다시 send → 중복