SQL에서 PII 컬럼 사용 추적

-- 개인정보 컬럼
ALTER TABLE catalog.schema.customers 
  ALTER COLUMN 이름 SET TAGS ('pii' = 'true', 'pii_type' = 'name');
ALTER TABLE catalog.schema.customers 
  ALTER COLUMN 전화번호 SET TAGS ('pii' = 'true', 'pii_type' = 'phone');
ALTER TABLE catalog.schema.customers 
  ALTER COLUMN 주민번호 SET TAGS ('pii' = 'true', 'pii_type' = 'ssn');

-- 익명/일반 컬럼
ALTER TABLE catalog.schema.customers 
  ALTER COLUMN 지역코드 SET TAGS ('pii' = 'false');
ALTER TABLE catalog.schema.customers 
  ALTER COLUMN 가입년월 SET TAGS ('pii' = 'false');

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# PII 컬럼 목록 정의
pii_columns = {
    "catalog.schema.customers": {
        "이름": "name", "전화번호": "phone", "주민번호": "ssn",
        "이메일": "email", "주소": "address"
    },
    "catalog.schema.orders": {
        "배송주소": "address", "연락처": "phone"
    }
}

for table, columns in pii_columns.items():
    for col, pii_type in columns.items():
        w.statement_execution.execute_statement(
            warehouse_id="<warehouse_id>",
            statement=f"ALTER TABLE {table} ALTER COLUMN {col} SET TAGS ('pii' = 'true', 'pii_type' = '{pii_type}')"
        )

-- 테이블의 컬럼별 PII 여부 확인
SELECT 
  column_name,
  tag_value AS is_pii
FROM system.information_schema.column_tags
WHERE catalog_name = 'catalog'
  AND schema_name = 'schema'
  AND table_name = 'customers'
  AND tag_name = 'pii'
ORDER BY column_name;

-- 고객이 이 쿼리를 실행했다고 가정
SELECT func(a), func(b)
FROM (
    SELECT 이름 AS a, 지역코드 AS b
    FROM catalog.schema.customers
);

-- 최근 쿼리에서 사용된 PII 컬럼 확인
SELECT 
  cl.source_table_full_name,
  cl.source_column_name,
  COALESCE(ct.tag_value, 'untagged') AS pii_status
FROM system.access.column_lineage cl
LEFT JOIN system.information_schema.column_tags ct
  ON cl.source_table_catalog = ct.catalog_name
  AND cl.source_table_schema = ct.schema_name
  AND cl.source_table_name = ct.table_name
  AND cl.source_column_name = ct.column_name
  AND ct.tag_name = 'pii'
WHERE cl.source_table_full_name = 'catalog.schema.customers'
  AND cl.event_time > current_timestamp() - INTERVAL 1 HOUR
ORDER BY cl.event_time DESC;

CREATE OR REPLACE FUNCTION catalog.schema.check_pii_columns(
  table_name STRING,
  column_list ARRAY<STRING>
)
RETURNS TABLE (column_name STRING, is_pii STRING, pii_type STRING)
RETURN
  SELECT 
    c.value AS column_name,
    COALESCE(ct.tag_value, 'unknown') AS is_pii,
    COALESCE(ct2.tag_value, '-') AS pii_type
  FROM EXPLODE(column_list) AS c
  LEFT JOIN system.information_schema.column_tags ct
    ON ct.table_name = split(table_name, '.')[2]
    AND ct.column_name = c.value
    AND ct.tag_name = 'pii'
  LEFT JOIN system.information_schema.column_tags ct2
    ON ct2.table_name = split(table_name, '.')[2]
    AND ct2.column_name = c.value
    AND ct2.tag_name = 'pii_type';

-- 사용 예시
SELECT * FROM catalog.schema.check_pii_columns(
  'catalog.schema.customers',
  ARRAY('이름', '지역코드', '전화번호')
);

-- PII 마스킹 함수
CREATE OR REPLACE FUNCTION catalog.schema.mask_pii(val STRING)
RETURNS STRING
RETURN CASE 
  WHEN is_member('pii_viewers') THEN val
  ELSE CONCAT(LEFT(val, 1), '***')
END;

-- 컬럼에 마스킹 적용
ALTER TABLE catalog.schema.customers 
  ALTER COLUMN 이름 SET MASK catalog.schema.mask_pii;
ALTER TABLE catalog.schema.customers 
  ALTER COLUMN 전화번호 SET MASK catalog.schema.mask_pii;