4. Data Decoration: Metadata Decorator#

1
from pyspark.sql import functions as F
2

3
# Kafka 레코드에 header로 메타데이터 추가
4
visits_with_metadata = (
5
    visits
6
    .withColumn("headers", F.array(
7
        F.struct(
8
            F.lit("job_version").alias("key"),
9
            F.lit(b"v1.2.3").alias("value")
10
        ),
11
        F.struct(
12
            F.lit("batch_version").alias("key"),
13
            F.lit(str(batch_number).encode("UTF-8")).alias("value")
14
        )
15
    ))
16
)
17

18
(visits_with_metadata.write.format("kafka")
19
    .option("kafka.bootstrap.servers", "localhost:9092")
20
    .option("includeHeaders", True)  # header 포함 명시
21
    .option("topic", "visits-output")
22
    .save())

1
-- 메타데이터 전용 테이블 (기술팀만 접근)
2
CREATE TABLE visits_context (
3
    execution_date_time TIMESTAMPTZ NOT NULL,
4
    loading_time        TIMESTAMPTZ NOT NULL,
5
    code_version        VARCHAR(15) NOT NULL,
6
    loading_attempt     SMALLINT NOT NULL,
7
    PRIMARY KEY (execution_date_time)
8
);
9

10
-- 데이터 테이블에 FK만 추가
11
-- visits 테이블의 visits_context_execution_date 컬럼은
12
-- view나 권한으로 비즈니스 사용자에게 숨김

1
-- 1) 메타데이터 삽입
2
INSERT INTO visits_context
3
  (execution_date_time, loading_time, code_version, loading_attempt)
4
VALUES (
5
  '{{ logical_date }}',
6
  '{{ dag_run.start_date }}',
7
  '{{ params.code_version }}',
8
  {{ task_instance.try_number }}
9
);
10

11
-- 2) 데이터 + 메타데이터 FK 삽입
12
INSERT INTO weekly_visits
13
  SELECT *, '{{ execution_date }}' AS visits_context_execution_date
14
  FROM tmp_visits;