[DRE] 실패 알림을 넘어 영향 범위 예측으로 — 데이터 파이프라인 Observability 구현 (with. Databricks)

📌 이 글의 핵심 요약 (Key Takeaways)

단순히 "작업이 실패했습니다"라고 통보하는 1차원적 모니터링을 넘어, 장애 발생 시 "어떤 후속 작업과 데이터 마트 테이블에 영향이 가는가?"를 예측하는 관측 가능성(Observability) 시스템 구축 실무 사례입니다. 본 글에서는 Databricks 파이프라인의 코드를 스캔하여 데이터 리니지(Lineage) 메타데이터를 자동화하는 방법, 다단계(A➡️B➡️C) 영향도 추적 로직, 그리고 알림 피로도(Alert Fatigue)를 방지하기 위해 MS Teams 채널을 세분화하여 공지하는 DRE(Data Reliability Engineering) 조직 문화 정착 노하우를 집중적으로 다룹니다.

---

 

안녕하세요. 주형권입니다. 

어느덧 Data Reliability Engineering를 도전한 지 7개월의 시간이 흘렀습니다. 단순히 모니터링을 넘어서 이제는 Observability라는 이름으로 무언가를 만들고 싶다는 생각이 들어서 단순히 1차원적 알림에서 나아가 어떠한 데이터에 영향이 갈지도 예측하는 방향으로 모니터링 시스템을 진화시키고자 합니다. 

 

이 글의 경우 어쩌면 기존에 작성 하였던 "[DRE] 데이터 파이프라인 장애 감지 자동화 — Airflow + Teams 연동"의 Step 2가 아닐까 싶습니다. 기존에는 단순히 이 작업이 "실패 하였으니 확인을 해봐"였다면 이제는 어떠한 "테이블과 작업이 영향이 있다"까지 보여주는 알림으로 진화시켰습니다.

 

[DRE] 데이터 파이프라인 장애 감지 자동화 — Airflow + Teams 연동

 

---

1. 무엇이 변경되었는가?

기존에는 앞서 말한 것과 같이 단순히 어떠한 작업이 실패하였는지 알림을 보냈습니다. 다음과 같은 형태로 알림이 발생했습니다.

담당자를 호출하여 본인의 작업이 실패하였다는 것을 인지 시키고, 간략하게 실패 원인에 대해서 보여줍니다. 사용자는 해당 파이프라인의 중요도에 따라 즉시 처리할지 후에 처리할지를 판단 가능하며, 자세한 내용을 보려면 View Run Details를 이용해서 작업으로 하이퍼링크를 통해서 확인도 가능합니다.

 

하지만 여기에서 한 가지 문제가 있습니다. 사용자가 만약 그 시스템에서 모든 것을 알고 있다면, 앞서 말한 판단을 할 때 망설임이 없이 즉시 처리할지 후에 처리할지를 판단할 수 있습니다. 그런데 여기에서 그 판단을 하려면 시스템을 모두 알아야합니다. 그리고 내가 만든 작업이 대규모 마트 작업이라서, 이로 인한 파생되는 테이블이 매우 많아지면 어느 부분이 문제가 되는지 알기 어렵습니다. 

 

그렇기 때문에 이를 판단 할 수 있도록 더욱 많은 정보가 필요합니다. 제가 여기에서 가장 필요하다고 느끼는 건 내 작업이 실패하면 어떠한 작업과 테이블에 영향이 가는가? 를 알 수 있어야 한다는 것입니다.

내 작업이 실패하면 어느 테이블 어느 작업까지 영향이 가는가?

 

이게 제가 생각하기에 Monitoring과 Observability의 차이라고 보입니다. 단순히 알렸다에서 무엇이 영향이 간다까지 알려주는 부분에서 기존의 작업 실패 알림을 넘어서 예측까지 하는 범위로 진화하였습니다. 

---

2. 설계와 구현

일단 첫 번째로 제일 중요한 것은 Source 테이블과 Target 테이블을 한눈에 알 수 있어야 한다는 것입니다. 그래야지 체인으로 모두 연결하여, A->B->C 이렇게 이어지는 파이프라인 체인을 통해서 어디서 잘못되면 어디서부터 어디까지 영향이 있는지를 정확히 알 수 있습니다. 

 

그래서 필요한 게 이러한 Source / Target 테이블의 정보와 해당 작업에서 어디서 데이터를 읽어서 어디로 적재하는지 정보를 담고 있는 정보가 필요하였습니다. 그래서 task_table_mapping이라는 테이블을 만들어서 Source와 Target 테이블의 정보 그리고 어느 작업에 영향을 받는지 모두 한눈에 조회가 가능한 테이블을 만들었습니다.

 

 

모니터링 시스템의 RDS에 저장하는 DDL을 다음과 같이 만들 수 있습니다. 어디까지나 DDL은 참고용이므로, VARCHAR 길이등은 너무 타이트하게 보지 마시고, 각자의 시스템에 맞게 조정을 부탁드립니다. (PK도 길이가 긴데, 어차피 디멘젼 느낌이라서 이렇게 잡는 게 확실하다고 판단하였습니다.)

CREATE TABLE reliability.task_table_mapping (
  job_id VARCHAR(100) NOT NULL COMMENT 'Job ID',
  job_name VARCHAR(255) NOT NULL COMMENT 'Job 이름',
  schedule VARCHAR(100) COMMENT '실행 스케줄',
  task_key VARCHAR(150) NOT NULL COMMENT 'Task 키',
  task_order INT NOT NULL COMMENT 'Task 실행 순서 (depends_on 기반)',
  notebook_path VARCHAR(500) COMMENT '노트북 경로',
  source_table VARCHAR(255) NOT NULL COMMENT '소스 테이블 (fully qualified)',
  target_table VARCHAR(255) NOT NULL COMMENT '타겟 테이블 (fully qualified)',
  write_method VARCHAR(100) COMMENT '적재 방식 (MERGE, CREATE OR REPLACE, INSERT, replaceWhere)',
  source_type VARCHAR(50) COMMENT '소스 유형 (UC_TABLE, S3, RDS)',
  
  -- 데이터가 UPDATE 될 때도 시간이 자동 변경되도록 트리거 추가
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '메타데이터 갱신 시각',
  
  -- 복합 기본키 지정 (총 길이: (100 + 150 + 200 + 200) * 4 = 2600 Bytes로 안전 범위 내 안착)
  PRIMARY KEY (job_id, task_key, source_table(200), target_table(200))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
  COMMENT='코드 스캔 기반 Job Task → Source/Target 테이블 매핑. 파이프라인 실패 시 영향 범위 파악 용도.';

.

 

처음에는 Databricks에서 제공하는 system.access.table_lineage / system.access.column_lineage을 사용하려고 하였으나 일회성으로 만든 테이블로 인해서 혼선을 줄 수 있다는 점과 완벽하게 커버가 불가능한다는 판단하에 실제로 Databricks의 작업에 있는 Task를 모두 일 1회 SCAN(추후에는 변경 부분만 SCAN으로 변경)하여, 실제로 등록된 Notebook / SQL 등을 읽어서 정규식 표현으로 Source / Target을 주기적으로 가져와서 reliability.task_table_mapping이라는 매핑 테이블에 적재하도록 하는 작업을 만들었습니다. (databricks의 Genie 코드의 도움을 받았으며, 아주 잘 가져옵니다.)

 

이렇게 하면 빠짐없이 가능하며, 실제로 주기적 작업에 등록된 작업의 Source / Target 정보 이외에 작업 정보와 어디로부터 오는지 여러 가지 정보를 얻을 수 있습니다.

def extract_tables_from_code(code):
    """코드에서 source/target 테이블을 정규식으로 추출한다."""
    if not code:
        return []
    
    results = []
    
    # --- Target 패턴 ---
    merge_targets = re.findall(
        r'MERGE\s+INTO\s+([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)',
        code, re.IGNORECASE
    )
    insert_targets = re.findall(
        r'INSERT\s+(?:INTO|OVERWRITE)\s+(?:TABLE\s+)?([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)',
        code, re.IGNORECASE
    )
    ctas_targets = re.findall(
        r'CREATE\s+(?:OR\s+REPLACE\s+)?TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)',
        code, re.IGNORECASE
    )
    save_targets = re.findall(
        r'\.saveAsTable\(\s*["\']([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)["\']',
        code, re.IGNORECASE
    )
    delete_targets = re.findall(
        r'DELETE\s+FROM\s+([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)',
        code, re.IGNORECASE
    )
    
    # --- Source 패턴 ---
    from_sources = re.findall(
        r'FROM\s+([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)',
        code, re.IGNORECASE
    )
    join_sources = re.findall(
        r'JOIN\s+([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)',
        code, re.IGNORECASE
    )
    spark_sources = re.findall(
        r'(?:spark\.table|spark\.read\.table)\(\s*["\']([a-zA-Z0-9_]+\.[a-zA-Z0-9_]+\.[a-zA-Z0-9_]+)["\']',
        code, re.IGNORECASE
    )
    s3_sources = re.findall(
        r'(s3://[a-zA-Z0-9\-_]+/[^\s"\'\)]+)',
        code
    )
    
    all_sources = list(set(from_sources + join_sources + spark_sources))
    all_targets = set()
    
    # Write method 판별 및 target 수집
    for t in merge_targets:
        all_targets.add((t, "MERGE"))
    for t in insert_targets:
        all_targets.add((t, "INSERT"))
    for t in ctas_targets:
        all_targets.add((t, "CREATE OR REPLACE"))
    for t in save_targets:
        all_targets.add((t, "saveAsTable"))
    for t in delete_targets:
        all_targets.add((t, "DELETE"))
    
    # replaceWhere 감지
    if re.search(r'replaceWhere', code, re.IGNORECASE):
        updated = set()
        for t, m in all_targets:
            if m in ("saveAsTable", "CREATE OR REPLACE"):
                updated.add((t, "replaceWhere"))
            else:
                updated.add((t, m))
        all_targets = updated
    
    # Source에서 target 제거 (self-reference 방지)
    target_names = {t for t, _ in all_targets}
    all_sources = [s for s in all_sources if s not in target_names]
    
    # Source type 판별 + 매핑 생성
    for target, method in all_targets:
        for source in all_sources:
            source_type = "UC_TABLE"
            if source.startswith("monitoring_rds"):
                source_type = "RDS"
            results.append((source, target, method, source_type))
        
        # S3 소스 추가 (날짜 변수 정규화)
        for s3 in s3_sources:
            s3_normalized = re.sub(r'/\{[^}]+\}', '/', s3)
            s3_normalized = re.sub(r'/\$\{[^}]+\}', '/', s3_normalized)
            s3_normalized = re.sub(r'/f"[^"]*"', '/', s3_normalized)
            s3_normalized = s3_normalized.rstrip('/')
            # 날짜 path 부분 제거 (/{year}/{month}/{day} 패턴)
            s3_normalized = re.sub(r'(/\d{4})?(/\d{2})?(/\d{2})?/?$', '/', s3_normalized).rstrip('/')
            results.append((s3_normalized, target, method, "S3"))
    
    return results

 

그래서 결론적으로 다음과 같이 기존에 성공/실패에 대한 알림만 전달하는 방식으로써 사용자에게 무엇이 실패했는지만 알림을 보냈다면 위에서 추가한 reliability.task_table_mapping 테이블을 통해서 예측을 하도록 구현합니다.

단순 상태 체크

 

다음과 같이 reliability.task_table_mapping을 통해서 영향도를 추가하여 알림을 보내도록 만들 수 있습니다. 단순히 테이블이 하나가 추가된 것 하나로 성공/실패뿐 아니라 영향도가 어디까지 있는지 알림을 보냅니다.

매핑 테이블을 이용한 영향도 체크

 

---

3. 알림과 후속 처리

기존에 단순히 성공/실패 여부만을 알려줬었다면, 이제는 진화하여 영향도까지 알 수 있도록 알림을 구성하였습니다.

(샘플 데이터입니다.)

진화된 알림

 

어떠한 작업이 실패하였으며, 이 작업이 실패함으로 인해서 다음의 테이블이 영향이 가는지 알 수 있습니다. 최대한 가독성이 좋게 하기 위해서 불필요한 메시지를 최대한 자제하려고 노력하였으며, 작업 실패로 인한 1차 테이블 그 1차 테이블을 바라보는 2차 테이블까지 표기가 됩니다. 물론 체인이 길게 있다면 1차 -> 2차 -> 3차 -> N차로 이어지게 됩니다. 이로써 담당자는 결정을 내리는데 더욱 많은 정보를 알기 쉽고 빠른 판단을 내릴 수 있습니다. 

 

하지만 여기에서 문제가 있습니다. 일단 알림을 보냈고, 작업의 담당자가 알림을 받아서 알았지만 이 테이블을 사용하는 사람들에게는 전파를 할 방법이 없습니다. 담당자가 내가 만든 테이블이 아닌 이걸 참조하는 테이블의 담당자까지 알기는 어렵습니다. 처음에는 system.query.history 또는 aduit을 이용해서 최근 30일 이내에 해당 테이블을 조회한 사용자를 역 추적해서 알림을 보내려고 하였지만 제가 늘 언급하듯이 배보다 배꼽이 큰 상황이 생길 거 같아서, 이 부분은 접었습니다. 

 

그리고 사람들이 습관적으로 SELECT * FROM 테이블 이런 식으로 데이터를 조회해서 샘플을 보고 판단하는 경우도 빈번하기에 실제 해당 사용자가 저 테이블을 사용하고 있다고 보기에는 확실히 무리가 있었습니다.

 

만약에 정말 큰 테이블의 작업이 장애가 난 경우 수백 명에서 하나씩 멘션을 줄수도 없고, 이런 메시지가 너무 빈번하다면 사람들은 피로도를 느껴서 정작 중요한 메시지를 놓치기 십상입니다. 그래서 아예 Teams에서 채널을 하나 만들어서 사람들이 이 채널을 통해서 어떠한 테이블에 문제가 있는지를 알 수 있도록 하는 것이 더욱 현실적인 방법이라고 생각했습니다.

Teams 채널을 통해서 전파

데이터 영향도 알림

 

초반에는 사람들이 우왕좌왕할 수 있으나 이 문화가 정착되면 자연스레 사람들은 이 채널을 통해서 장애 여부를 알 수 있을 것입니다. 또한 해당 메시지에 다음과 같이 댓글을 이용해서 담당자가 언제 복구되었는지 얼마나 소요될지에 대해서도 알 수 있습니다. 물론 SLI / SLO / SLA 등을 확인하기 위해서 복구를 자동으로 감지하도록 할 수도 있지만 현재 DRE의 문화가 정착이 안된 시점에서 너무 이른 판단으로 보여, 다음과 같이 후속 처리는 수동으로 하였습니다.

 

자문자답 상황극...

 

위와 같이 소소하지만 정확하게 담당자가 처리 이후에 댓글을 달아서 해당 작업이 잘 처리되었다고 다시금 공지를 할 수 있으며, 사람들은 당연히 해당 공지를 통해서 자신의 후속 작업이나 새로운 데이터 작업을 완료할 수 있습니다. 

---

마치며

지난 7개월간 Data Reliability Engineering(DRE) 업무를 도전하며 얻은 가장 큰 교훈은, "데이터 신뢰성은 완벽한 기술력만으로 완성되지 않는다"는 것입니다. 아무리 정교한 모니터링 시스템과 리니지(Lineage) 추적 기술을 도입하더라도, 장애 상황을 투명하게 공유하고 장애 복구에 대해 적극적으로 소통하는 '문화'가 정착되지 않는다면 그 시스템은 결국 외면받게 됩니다.

 

이번에 구축한 '영향도 예측 알림'과 '이상탐지 전용 공지 채널' 역시, 엔지니어와 현업 사용자가 서로의 상황을 이해하고 신뢰할 수 있도록 돕는 소통의 다리가 되기를 바라는 마음에서 시작되었습니다.

 

아직 우리 조직의 DRE 문화가 완전히 무르익었다고 할 수는 없지만, 이렇게 작지만 확실한 시스템적 변화들이 모여 단단한 데이터 문화를 만들어갈 것이라 확신합니다. 앞으로도 단순한 알림을 넘어, 모두가 믿고 쓸 수 있는 진정한 의미의 '관측 가능성(Observability)'을 구현하기 위해 끊임없이 고민하고 발전시켜 나가겠습니다.

 

긴 글 읽어 주셔서 감사합니다.