1
from pyspark.sql import SparkSession
2

3
def passthrough_replicate_files(spark: SparkSession, source: str, dest: str):
4
    """
5
    Passthrough Replicator — 가장 단순한 텍스트 API 사용.
6
    JSON I/O API가 아닌 text API로 행을 그대로 복사.
7
    """
8
    # text API → 파싱 없이 라인 그대로 복사
9
    input_dataset = spark.read.text(source)
10
    input_dataset.write.mode("overwrite").text(dest)

1
[Pull 방식] — 위험
2
staging 환경이 production DB에 접근하여 데이터를 가져옴
3
→ staging의 버그가 production에 영향 줄 수 있음.ex) IP 고갈
4

5
[Push 방식] — 권장
6
production 환경이 직접 staging으로 데이터를 복사
7
→ production이 프로세스를 통제 (빈도, 처리량 제어)

1
# Push 방식 구현 개념
2
def push_replicate(spark: SparkSession, prod_path: str, target_envs: list[str]):
3
    """
4
    Push 방식: production이 다른 환경으로 데이터를 밀어넣음.
5
    production이 빈도와 처리량을 통제.
6
    """
7
    source_data = spark.read.text(prod_path)
8

9
    for env_path in target_envs:
10
        source_data.write.mode("overwrite").text(env_path)
11

12
# push_replicate(spark, "s3://prod/devices", ["s3://staging/devices", "s3://dev/devices"])

1
def replicate_kafka_with_ordering(events_stream, output_topic: str, checkpoint: str):
2
    from pyspark.sql import DataFrame
3

4
    events_to_replicate = events_stream.selectExpr(
5
        "key", "value", "partition", "headers", "offset"
6
    )
7

8
    def write_sorted_events(events: DataFrame, batch_id: int):
9
        (events
10
         .sortWithinPartitions("offset", ascending=True)  # 순서보존핵심
11
         .drop("offset")
12
         .write.format("kafka")
13
         .option("kafka.bootstrap.servers", "broker:9092")
14
         .option("topic", output_topic)
15
         .option("includeHeaders", "true")  # 헤더
16
         .save())
17

18
    (events_to_replicate.writeStream
19
     .option("checkpointLocation", checkpoint)
20
     .foreachBatch(write_sorted_events)
21
     .start())

1
s3_replication_config = {
2
    "role": "arn:aws:iam::role/replication",
3
    "source_bucket": "devices-production",
4
    "rules": [{
5
        "id": "devices",
6
        "status": "Enabled",
7
        "destination": {
8
            "bucket": "devices-staging",
9
            "storage_class": "STANDARD"
10
        }
11
    }]
12
}
13
# 실제로는 Terraform HCL이나 AWS CLI로 설정

1
def replicate_with_column_drop(spark: SparkSession, source: str, dest: str):
2
    dataset = spark.read.format("delta").load(source)
3

4
    safe_dataset = dataset.drop("ip", "latitude", "longitude")
5
    safe_dataset.write.mode("overwrite").format("delta").save(dest)

1
from pyspark.sql import functions as F
2

3
def replicate_with_anonymization(spark: SparkSession, source: str, dest: str):
4
    """PII 컬럼을 변환(난독화)하고 복제"""
5
    dataset = spark.read.format("delta").load(source)
6

7
    anonymized = dataset.withColumn(
8
        "full_name",
9
        F.expr("SUBSTRING(full_name, 2, LENGTH(full_name))")  # 첫 글자 제거
10
    )
11

12
    anonymized.write.mode("overwrite").format("delta").save(dest)

1
grant_sql = """
2
    GRANT SELECT (visit_id, event_time, user_id)
3
    ON TABLE visits
4
    TO user_a
5
"""
6
# user_a는 ip, latitude, longitude에 접근 불가
7
# 복제 잡이 user_a 권한으로 실행되면 자동으로 PII 제외

1
# 위험: timestamp 파싱 시 포맷 불일치로 null 변환
2
schema_risky = "id STRING, event_time TIMESTAMP, name STRING"
3

4
# 안전: timestamp을 string으로 유지
5
schema_safe = "id STRING, event_time STRING, name STRING"

1
[수동 관리] — 위험
2
변환 로직에 ip, latitude, longitude 하드코딩
3
→ 새 PII 컬럼(phone_number) 추가 시 누락 가능
4

5
[자동화] — 권장
6
Data Catalog에서 PII 태그된 컬럼 목록을 동적으로 조회
7
→ 변환 로직이 항상 최신 PII 정의를 반영