#!/usr/bin/env python3 import subprocess import sqlparse from sqlparse.sql import Identifier, IdentifierList # 설정 IMPALA_HOST = "your_impala_host" IMPALA_PORT = "21050" DATABASE = "your_database" OUTPUT_FILE = "view_table_mapping.txt" def run_impala_query(query): cmd = f"impala-shell -i {IMPALA_HOST}:{IMPALA_PORT} -d {DATABASE} -q '{query}' --quiet -B" result = subprocess.run(cmd, shell=True, capture_output=True, text=True) return result.stdout.strip() def extract_tables(ddl): parsed = sqlparse.parse(ddl)[0] tables = set() for token in parsed.tokens: if isinstance(token, (Identifier, IdentifierList)): if token.get_parent_name(): # db.table 형식 tables.add(f"{token.get_parent_name()}.{token.get_real_name()}") else: tables.add(token.get_real_name()) return tables # 메인 로직 with open(OUTPUT_FILE, 'w') as f: f.write("View와 물리 테이블 매핑 정보\n") f.write(f"실행 시간: {import_time.strftime('%Y-%m-%d %H:%M:%S')}\n") f.write("----------------------------------------\n") views = run_impala_query("SHOW VIEWS;").splitlines() if not views: f.write(f"데이터베이스 {DATABASE}에 View가 존재하지 않습니다.\n") exit(0) for view in views: ddl = run_impala_query(f"SHOW CREATE VIEW {view};") tables = extract_tables(ddl) f.write(f"View: {view}\n") if tables: f.write("Physical Tables:\n") for table in sorted(tables): f.write(f" - {table}\n") else: f.write("Physical Tables: (추출 실패)\n") f.write("----------------------------------------\n") print(f"처리 완료. 결과는 {OUTPUT_FILE}에 저장되었습니다.")