- 데이터인사이트

리스트

마이리스트

마이리뷰

마이리뷰

마이페이퍼

마이페이퍼


방명록


서재 태그

more



powered by aladin

북마크하기 [100자평] 이기적 경영정보시각화능력 필기 기본서ｌ마이리뷰

댓글(3)

데이터인사이트 l 2024-03-19 09:30

https://blog.aladin.co.kr/733595173/15392112

이기적 경영정보시각화능력 필기 기본서 - 최신판 + 신설 국가기술자격증 + CBT 온라인 문제집
인천대학교 비즈니스데이터센터.경기대 빅데이터 시각화 연구실.이베로니카 지음 / 영진.com(영진닷컴) / 2024년 3월
평점 :
구판절판


오랫동안 OLAP 시스템을 운영하면서 태블로 등의 자격증에 관심을 가지고 있었으나 해외 자격증이라 미루고 있었는데 국내 자격증이 생겨서 관심을 가지게 되었네요. 믿을만한 "이기적"시리즈에서 출간되어 기대됩니다

댓글(3) 먼댓글(0) 좋아요(0)
좋아요
ｌ 공유하기 ｌ 북마크하기찜하기 ｌthankstoThanksTo
먼댓글 주소 : https://blog.aladin.co.kr/trackback/733595173/15392112
주소복사 ㅣ

 
 
데이터인사이트 2026-10-30 11:03   좋아요 0 | 댓글달기 | URL
import io
import os
import re
import sys
import textwrap
from contextlib import contextmanager, redirect_stdout
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Optional

from pyspark.sql import SparkSession


# ==================== 실행 설정 ====================
VALIDATE_QUERY_FILES = [
˝.sql˝,
]

OUTPUT_FILE_NAME = ˝.txt˝ # 결과 저장 파일명
PREVIEW_ROWS = 40 # 결과 있는 쿼리의 미리보기 행 수
KST = timezone(timedelta(hours=9))

# 결과 출력 방식
# ˝immediate˝ : 쿼리 실행 직후 결과/오류를 바로 출력
# ˝deferred˝ : 전체 실행이 끝난 뒤 0건 -> 결과 있음 -> 오류 순서로 한꺼번에 출력
RESULT_OUTPUT_MODE = ˝immediate˝

IMMEDIATE = ˝immediate˝
DEFERRED = ˝deferred˝

# SQL 파일 내 ${변수명}에 치환될 값
SQL_VARIABLES = {
˝plant_no˝: ˝˝,
˝schema˝: ˝˝,
˝ldg_ymd˝: ˝2026-10-04˝,
˝DW_LDG_YMD˝: ˝9999-12-31˝,

˝LOAD_JOB_GUBUN˝: ˝‘‘˝, # 대체값: ˝table_cnt˝
˝LOAD_TGT_TAB˝: ˝‘‘˝,
˝LOAD_PLANT_NO˝: ˝--and = ‘0003‘˝,

˝LOAD_JOB_GUBUN2˝: ˝and = ‘table_cnt‘˝,
# 대체값: ˝and LOAD_SEQ in (‘4464‘,‘10139‘,‘10140‘)˝
˝LOAD_TGT_TAB2˝: ˝and LOAD_TGT_TAB = ‘‘˝,
˝LOAD_PLANT_NO2˝: ˝and LOAD_PLANT_NO = ‘‘˝,

˝WHERE˝: ˝WHERE 1=1˝, # 대체값: ˝WHERE tgt_cnt <> sum_by_src_col_val˝
}

# 실행 전에 REFRESH TABLE 할 대상
REFRESH_TABLES = [
˝a_dsp_mfr.˝,
˝a_dsp_mfr.˝,
˝a_dsp_mfr.˝,
˝a_dsp_mfr.˝,
]

# Spark 세션 설정
SPARK_CONF = {
˝spark.sql.extensions˝: ˝org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions˝,
˝spark.sql.catalog.spark_catalog˝: ˝org.apache.iceberg.spark.SparkSessionCatalog˝,
˝spark.sql.iceberg.handle-timestamp-without-timezone˝: ˝true˝,
˝spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive˝: ˝true˝,
˝spark.sql.autoBroadcastJoinThreshold˝: ˝-1˝,
˝spark.sql.adaptive.autoBroadcastJoinThreshold˝: ˝-1˝,
˝spark.sql.adaptive.enabled˝: ˝false˝,
˝spark.sql.shuffle.partitions˝: ˝800˝,
}


# ==================== 결과 자료구조 ====================
@dataclass
class QueryResult:
file: str
query_no: int
sql: str
sql_head: str # SQL 앞 3줄 (오류 헤더용)
row_count: int = 0
preview: str = ˝˝
error: Optional[str] = None

@property
def ok(self) -> bool:
return self.error is None


# ==================== 출력 유틸 (화면 + 파일 동시 출력) ====================
class TeeWriter:
˝˝˝write 호출을 원래 stdout과 파일에 동시에 기록한다.˝˝˝

def __init__(self, stream, file):
self.stream = stream
self.file = file

def write(self, data: str) -> int:
self.stream.write(data)
self.file.write(data)
self.file.flush() # 강제 종료되어도 출력이 남도록 매번 flush
return len(data)

def flush(self):
self.stream.flush()
self.file.flush()


@contextmanager
def tee_stdout(file_path: str):
˝˝˝with 블록 안의 print 출력을 화면과 파일에 함께 남기고, 종료 시 stdout을 복원한다.˝˝˝
with open(file_path, ˝w˝, encoding=˝utf-8˝) as f:
with redirect_stdout(TeeWriter(sys.stdout, f)):
yield


def print_banner(title: str, width: int = 180) -> None:
print(f˝n{‘=‘ * width}˝)
print(title)
print(f˝{‘=‘ * width}˝)


def print_sub_banner(title: str, width: int = 80) -> None:
print(f˝n{‘-‘ * width}˝)
print(title)
print(f˝{‘-‘ * width}˝)


def now_str() -> str:
return datetime.now(KST).strftime(˝%Y%m%d_%H%M%S˝)


# ==================== 문자열/SQL 유틸 ====================
def replace_sql_variables(sql_text: str, variables: dict) -> str:
˝˝˝
SQL 내부의 ${변수명}을 variables 값으로 치환한다.
정의되지 않은 변수가 있으면 ValueError를 발생시킨다.
˝˝˝
def _replace(match):
name = match.group(1)
if name not in variables:
available = ˝, ˝.join(variables.keys())
raise ValueError(
f˝정의되지 않은 SQL 변수: ‘${{{name}}}‘n사용 가능한 변수: {available}˝
)
return str(variables[name])

return re.sub(r˝${([a-zA-Z_][a-zA-Z0-9_]*)}˝, _replace, sql_text)


def remove_iceberg_prefix(sql_text: str) -> tuple:
˝˝˝
‘iceberg.‘ 접두어를 제거한다 (대소문자 무시).
예) FROM iceberg.a_dsp_mfr.DW_MODEL_B -> FROM a_dsp_mfr.DW_MODEL_B
Returns: (변환된 SQL, 제거 건수)
˝˝˝
pattern = re.compile(r˝(?<![A-Za-z0-9_])iceberg.˝, re.IGNORECASE)
return pattern.subn(˝˝, sql_text)


def split_statements(sql_text: str) -> list:
˝˝˝‘;‘ 기준으로 SQL 문장을 나누고 빈 문장은 제외한다.˝˝˝
return [s.strip() for s in sql_text.split(˝;˝) if s.strip()]


def first_lines(sql: str, n: int = 3) -> str:
˝˝˝SQL의 앞 n줄을 반환한다. 빈 줄은 제외하고 실제 내용이 있는 줄만 센다.˝˝˝
lines = [line.strip() for line in sql.splitlines() if line.strip()]
return ˝n˝.join(lines[:n])


def short_error(e: Exception, max_lines: int = 2) -> str:
˝˝˝오류 메시지의 앞 몇 줄만 사용한다.˝˝˝
lines = [line.strip() for line in str(e).splitlines() if line.strip()]
return ˝n˝.join(lines[:max_lines])


def validate_output_mode(mode: str) -> str:
if mode not in (IMMEDIATE, DEFERRED):
raise ValueError(
f˝RESULT_OUTPUT_MODE 값이 잘못됨: {mode!r} (허용: {IMMEDIATE!r}, {DEFERRED!r})˝
)
return mode

데이터인사이트 2026-10-30 11:00   좋아요 0 | 댓글달기 | URL

# ==================== Spark ====================
def build_spark() -> SparkSession:
˝˝˝Spark 세션을 만들고 설정을 적용한다.˝˝˝
builder = SparkSession.builder.appName(˝create integrate no˝)
for key, value in SPARK_CONF.items():
builder = builder.config(key, value)
spark = builder.getOrCreate()

spark.sparkContext._jsc.hadoopConfiguration().set(
˝mapreduce.input.fileinputformat.input.dir.recursive˝, ˝true˝
)
spark.sparkContext.setLogLevel(˝ERROR˝)
spark.catalog.clearCache()
return spark


def refresh_tables(spark: SparkSession) -> None:
˝˝˝설정된 테이블의 메타데이터를 갱신한다.˝˝˝
for table in REFRESH_TABLES:
spark.sql(f˝REFRESH TABLE {table}˝)


# ==================== SQL 파일 로드 ====================
def load_query_file(path: str) -> str:
˝˝˝
SQL 파일을 읽고 변수 치환과 iceberg. 제거를 적용한다.
파일 오류는 예외로 올라가며, 호출부에서 파일 단위 오류로 기록한다.
˝˝˝
with open(path, ˝r˝, encoding=˝utf-8˝) as f:
text = f.read()

text = replace_sql_variables(text, SQL_VARIABLES)
text, removed = remove_iceberg_prefix(text)
print(f˝[iceberg. 제거 건수] {removed}건˝)
return text


# ==================== 개별 결과 출력 ====================
def print_result_block(r: QueryResult) -> None:
˝˝˝성공한 쿼리 1건의 결과(SQL + 미리보기)를 출력한다.˝˝˝
print_sub_banner(f˝[{r.file}] 쿼리 {r.query_no} (결과 {r.row_count}행)˝)
#print(r.sql)
#print()
print(r.preview, end=˝˝)


def print_error_block(r: QueryResult) -> None:
˝˝˝실패한 쿼리 1건의 오류 내용과 SQL을 출력한다.˝˝˝
print_sub_banner(f˝[{r.file}] 쿼리 {r.query_no} 오류˝)
print(˝[SQL 앞 3줄]˝)
print(textwrap.indent(r.sql_head, ˝ ˝))
print(f˝[오류 내용] {r.error}˝)
print_sub_banner(f˝[{r.file}] 쿼리 {r.query_no} SQL˝)
print(r.sql)
print()


def print_query_outcome(r: QueryResult) -> None:
˝˝˝성공/실패에 따라 적절한 블록을 출력한다.˝˝˝
if r.ok:
print_result_block(r)
else:
print_error_block(r)


# ==================== 쿼리 실행 ====================
def run_statement(spark: SparkSession, file: str, query_no: int, sql: str,
show_now: bool) -> QueryResult:
˝˝˝
쿼리 1건을 실행하고 건수와 미리보기를 기록한다.
show_now가 True면 실행 직후 결과를 바로 출력한다.
오류가 나도 예외를 올리지 않고 다음 쿼리로 진행한다.
˝˝˝
result = QueryResult(file=file, query_no=query_no, sql=sql,
sql_head=first_lines(sql, 3))
print(f˝[실행 {query_no}] [SQL 전체]n{sql}˝)
#print(f˝[실행 {query_no}]˝)
try:
df = spark.sql(sql)
result.row_count = df.count()

if result.row_count > 0:
buf = io.StringIO()
with redirect_stdout(buf):
df.show(PREVIEW_ROWS, truncate=False, vertical=False)
result.preview = buf.getvalue()

print(f˝[성공 {query_no}] 데이터 {result.row_count}건˝)

except Exception as e:
result.error = short_error(e)
print(f˝[실패 {query_no}] {now_str()}˝)

if show_now:
print_query_outcome(result)

return result


def run_file(spark: SparkSession, file: str, file_idx: int, total: int,
mode: str) -> tuple:
˝˝˝
SQL 파일 1개를 실행한다.
Returns: (쿼리 결과 리스트, 파일 단위 오류 dict 또는 None)
˝˝˝
print_banner(f˝[조회 파일 {file_idx}/{total}] {file}˝)

try:
query_text = load_query_file(file)
except FileNotFoundError:
print(f˝❌ 파일을 찾을 수 없음: {file}˝)
return [], {˝file˝: file, ˝error˝: ˝File not found˝}
except ValueError as e:
print(f˝❌ SQL 변수 치환 실패: {file}˝)
return [], {˝file˝: file, ˝error˝: str(e)}

statements = split_statements(query_text)
show_now = (mode == IMMEDIATE)
results = [
run_statement(spark, file, no, sql, show_now)
for no, sql in enumerate(statements, start=1)
]

ok_count = sum(1 for r in results if r.ok)
print_sub_banner(f˝📌 파일 {file_idx} 결과 요약: {file}˝, width=80)
print(f˝전체 쿼리 수 : {len(statements)}˝)
print(f˝✅ 성공 수 : {ok_count}˝)
print(f˝❌ 실패 수 : {len(results) - ok_count}˝)

return results, None


# ==================== 최종 출력 ====================
def print_totals(all_results: list, file_count: int) -> None:
zero = [r for r in all_results if r.ok and r.row_count == 0]
nonzero = [r for r in all_results if r.ok and r.row_count > 0]
errors = [r for r in all_results if not r.ok]

print_banner(˝✅ 전체 조회 쿼리 처리 완료˝)
print(˝n📊 [조회 쿼리 최종 결과]˝)
print(f˝처리된 파일 수 : {file_count}˝)
print(f˝✅ 성공 쿼리 수 : {len(all_results) - len(errors)}˝)
print(f˝❌ 실패 쿼리 수 : {len(errors)}˝)
print(f˝📋 결과 0건 쿼리 수 : {len(zero)}˝)
print(f˝📋 결과 있는 쿼리 수 : {len(nonzero)}˝)


def print_deferred_results(all_results: list) -> None:
˝˝˝DEFERRED 모드: 0건 -> 결과 있음 -> 오류 순서로 한꺼번에 출력한다.˝˝˝
zero = [r for r in all_results if r.ok and r.row_count == 0]
nonzero = [r for r in all_results if r.ok and r.row_count > 0]
errors = [r for r in all_results if not r.ok]

print_banner(f˝📋 [결과 0건 쿼리 목록] : {len(zero)}건˝)
for r in zero:
print_result_block(r)

print_banner(f˝📋 [결과 있는 쿼리 상세] : {len(nonzero)}건˝)
for r in nonzero:
print_result_block(r)

print_banner(f˝❌ [쿼리 단위 오류 상세] : {len(errors)}건˝)
for r in errors:
print_error_block(r)


def print_immediate_notice(all_results: list) -> None:
˝˝˝IMMEDIATE 모드: 결과는 이미 출력되었으므로 분류 개수만 알려준다.˝˝˝
zero = sum(1 for r in all_results if r.ok and r.row_count == 0)
nonzero = sum(1 for r in all_results if r.ok and r.row_count > 0)
errors = sum(1 for r in all_results if not r.ok)

print_banner(˝📋 [결과 출력] 실행 중 즉시 출력 완료˝)
print(f˝- 결과 0건 : {zero}건˝)
print(f˝- 결과 있음 : {nonzero}건˝)
print(f˝- 오류 : {errors}건 (상세는 실행 중 출력됨)˝)


def print_error_summary(all_results: list, file_errors: list) -> None:
˝˝˝파일 단위 오류는 항상 출력한다. 쿼리 오류는 개수만 요약한다.˝˝˝
errors = [r for r in all_results if not r.ok]
print_banner(f˝❌ [오류 정리] : {len(file_errors) + len(errors)}건˝)

if file_errors:
print(f˝n[파일 단위 오류] : {len(file_errors)}건˝)
for fe in file_errors:
print(f˝- 파일: {fe[‘file‘]}˝)
print(f˝ 오류: {fe[‘error‘]}˝)

데이터인사이트 2026-10-30 11:00   좋아요 0 | 댓글달기 | URL

if errors:
print(f˝n[쿼리 단위 오류] : {len(errors)}건˝)
for r in errors:
print(f˝- [{r.file}] 쿼리 {r.query_no}: {r.error.splitlines()[0]}˝)


# ==================== 메인 ====================
def main() -> None:
mode = validate_output_mode(RESULT_OUTPUT_MODE)
spark = build_spark()

with tee_stdout(OUTPUT_FILE_NAME):
print(f˝✅ Work Dir: {os.getcwd()}˝)
print(f˝✅ Start Time: {now_str()}˝)
print(f˝✅ 결과 출력 방식: {mode}˝)

refresh_tables(spark)

print_banner(˝검증 쿼리 파일 반복 실행˝)

all_results = []
file_errors = []
for idx, file in enumerate(VALIDATE_QUERY_FILES, start=1):
results, file_error = run_file(spark, file, idx, len(VALIDATE_QUERY_FILES), mode)
all_results.extend(results)
if file_error:
file_errors.append(file_error)

print_totals(all_results, len(VALIDATE_QUERY_FILES))

if mode == DEFERRED:
print_deferred_results(all_results)
else:
print_immediate_notice(all_results)

print_error_summary(all_results, file_errors)

print(f˝n[전체종료] {now_str()}˝)
print(˝=˝ * 180)

print(f˝결과 파일 저장 완료: {OUTPUT_FILE_NAME}˝)


if __name__ == ˝__main__˝:
main()