- 데이터인사이트

리스트

마이리스트

마이리뷰

마이리뷰

마이페이퍼

마이페이퍼


방명록


서재 태그

more



powered by aladin

북마크하기 [100자평] 거인을 읽다ｌ마이리뷰

댓글(2)

데이터인사이트 l 2025-10-15 01:10

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

거인을 읽다
신영준.고영성 엮음 / 상상스퀘어 / 2025년 10월
평점 :
장바구니담기


유튜브에 광고를 하길래 뭔가 대단한 책인 줄 알고 검색 해 봤더니, 자기가 쓴 책을 광고하고 있다.
작가의 생각은 하나도 없고, 한 페이지에 3개의 명언을 짜깁기한 것을 책이라고 하다니... 말문이 막힌다.

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

 
 
데이터인사이트 2026-10-14 15:18   좋아요 0 | 댓글달기 | URL
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StringType, StructField, StructType
from datetime import datetime, timezone, timedelta
from pathlib import Path
import os
import re

# ==================== 조회 쿼리 파일 목록 정의 ====================
validate_query_files = [
˝z_insert_error3.sql˝
]


# ==================== SQL 변수 정의 ====================
plant_no = ˝˝
schema = ˝aaa˝
ldg_ymd = ˝2026-10-04˝

sql_variables = {
˝plant_no˝: plant_no,
˝schema˝: schema,
˝ldg_ymd˝: ldg_ymd
}


# ==================== 기본 환경 설정 ====================
KST = timezone(timedelta(hours=9))
work_dir = os.getcwd()


# ==================== Spark Session 생성 ====================
spark = SparkSession.builder
.appName(˝create integrate no˝)
.config(
˝spark.sql.extensions˝,
˝org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions˝
)
.config(
˝spark.sql.catalog.spark_catalog˝,
˝org.apache.iceberg.spark.SparkSessionCatalog˝
)
.config(
˝spark.sql.iceberg.handle-timestamp-without-timezone˝,
˝true˝
)
.getOrCreate()


# ==================== Spark 환경 설정 ====================
spark.conf.set(
˝spark.hadoop.mapreduce.input.fileinputformat.input.dir.recursive˝,
˝true˝
)

spark.sparkContext._jsc.hadoopConfiguration().set(
˝mapreduce.input.fileinputformat.input.dir.recursive˝,
˝true˝
)

spark.conf.set(˝spark.sql.autoBroadcastJoinThreshold˝, ˝-1˝)
spark.conf.set(˝spark.sql.adaptive.autoBroadcastJoinThreshold˝, ˝-1˝)
spark.conf.set(˝spark.sql.adaptive.enabled˝, ˝false˝)
spark.conf.set(˝spark.sql.shuffle.partitions˝, ˝800˝)

spark.sparkContext.setLogLevel(˝ERROR˝)


# ==================== 캐시 초기화 및 테이블 새로고침 ====================
spark.catalog.clearCache()

refresh_tables = [
˝aaa.bbb˝
]

for table_name in refresh_tables:
spark.sql(f˝REFRESH TABLE {table_name}˝)


print(f˝✅ Work Dir: {work_dir}˝)
print(f˝✅ Start Time: {datetime.now(KST).strftime(‘%Y%m%d_%H%M%S‘)}˝)




# ==================== SQL 변수 치환 함수 ====================
def replace_sql_variables(sql_text: str, variables: dict) -> str:
˝˝˝
SQL 내부의 ${변수명}을 variables 딕셔너리 값으로 치환합니다.

예시:
${plant_no} -> aaaa
${schema} -> aaaa
${ldg_ymd} -> 2026-10-04
˝˝˝

def replace_match(match):
variable_name = match.group(1)

if variable_name not in variables:
available_variables = ˝, ˝.join(variables.keys())

raise ValueError(
f˝정의되지 않은 SQL 변수: ‘${{{variable_name}}}‘n˝
f˝사용 가능한 변수: {available_variables}˝
)

return str(variables[variable_name])

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


# ==================== 전체 결과 저장 ====================
overall_validate_success = []
overall_validate_fail = []


# ==================== 조회 쿼리 파일 반복 처리 ====================
print(f˝n{‘=‘ * 180}˝)
print(˝검증 쿼리 파일 반복 실행˝)
print(f˝{‘=‘ * 180}˝)


for file_idx, validate_file_path in enumerate(validate_query_files, start=1):

print(f˝n{‘=‘ * 180}˝)
print(f˝[조회 파일 {file_idx}/{len(validate_query_files)}] {validate_file_path}˝)
print(f˝{‘=‘ * 180}˝)

# -------------------- SQL 파일 읽기 및 변수 치환 --------------------
try:
with open(validate_file_path, ˝r˝, encoding=˝utf-8˝) as f:
validate_query = f.read()

# SQL 파일의 ${plant_no}, ${schema}, ${ldg_ymd} 등 치환
validate_query = replace_sql_variables(
validate_query,
sql_variables
)

#print(˝n[치환 완료 SQL]˝)
#print(validate_query)

except FileNotFoundError:
error_message = f˝파일을 찾을 수 없음: {validate_file_path}˝

print(f˝❌ {error_message}˝)

overall_validate_fail.append({
˝file˝: validate_file_path,
˝query_no˝: None,
˝sql_first_line˝: None,
˝error˝: error_message
})

continue

except ValueError as e:
error_message = str(e)

print(f˝❌ SQL 변수 치환 실패: {validate_file_path}˝)
print(f˝[오류 내용]n{error_message}˝)

overall_validate_fail.append({
˝file˝: validate_file_path,
˝query_no˝: None,
˝sql_first_line˝: None,
˝error˝: error_message
})

continue

except Exception as e:
error_message = str(e)

print(f˝❌ SQL 파일 처리 중 알 수 없는 오류 발생: {validate_file_path}˝)
print(f˝[오류 내용]n{error_message}˝)

overall_validate_fail.append({
˝file˝: validate_file_path,
˝query_no˝: None,
˝sql_first_line˝: None,
˝error˝: error_message
})

continue


# -------------------- 파일별 성공/실패 결과 --------------------
file_validate_success = []
file_validate_fail = []


# -------------------- SQL을 세미콜론(;) 기준으로 분리 --------------------
sql_list = [
sql.strip()
for sql in validate_query.split(˝;˝)
if sql.strip()
]

데이터인사이트 2026-10-14 15:18   좋아요 0 | 댓글달기 | URL

# ==================== SQL 반복 실행 ====================
for i, sql in enumerate(sql_list, start=1):

# 빈 줄이 포함된 경우를 고려하여 첫 번째 유효 라인 추출
sql_lines = [
line.strip()
for line in sql.splitlines()
if line.strip()
]

sql_first_line = ˝n˝.join(sql_lines[:3]).strip() if sql_lines else ˝(빈 SQL)˝
try:
# -------------------- 실행 시작 로그 --------------------
print(f˝n{‘-‘ * 180}˝)
print(
f˝[실행 {i}/{len(sql_list)}] ˝
f˝{datetime.now(KST).strftime(‘%Y%m%d_%H%M%S‘)}˝
)
print(f˝{‘-‘ * 180}˝)
print(sql_first_line)
print()


# -------------------- SQL 실행 --------------------
result_df = spark.sql(sql)


# -------------------- 결과 출력 --------------------
# 전체 행 수가 필요하면 아래 주석을 해제하세요.
# row_count = result_df.count()

result_df.show(
40,
truncate=False,
vertical=False
)


# count()를 실행하지 않는 경우 출력 행 수 기준으로 기록
row_count = ˝show(20) 실행 완료˝


# -------------------- 성공 결과 저장 --------------------
file_validate_success.append({
˝file˝: validate_file_path,
˝query_no˝: i,
˝sql_first_line˝: sql_first_line,
˝row_count˝: row_count
})


except Exception as e:
error_message = str(e)

# -------------------- 실패 로그 --------------------
print(
f˝n❌ [실패 {i}/{len(sql_list)}] ˝
f˝{datetime.now(KST).strftime(‘%Y%m%d_%H%M%S‘)}˝
)
print(f˝[SQL 첫 줄] {sql_first_line}˝)
print(f˝[오류 내용]n{error_message}n˝)


# -------------------- 실패 결과 저장 --------------------
file_validate_fail.append({
˝file˝: validate_file_path,
˝query_no˝: i,
˝sql_first_line˝: sql_first_line,
˝error˝: error_message
})

# 오류가 발생해도 다음 SQL 계속 실행
continue


# ==================== 파일별 결과 출력 ====================
if file_validate_fail:
print(f˝n{‘=‘ * 180}˝)
print(f˝❌ [{validate_file_path}] 실패 쿼리 목록˝)
print(f˝{‘=‘ * 180}˝)

for fail in file_validate_fail:
print(f˝n- 쿼리 {fail[‘query_no‘]}: {fail[‘sql_first_line‘]}˝)
print(f˝ 오류: {fail[‘error‘]}˝)


if file_validate_success:
print(f˝n{‘=‘ * 180}˝)
print(f˝✅ [{validate_file_path}] 성공 쿼리 목록˝)
print(f˝{‘=‘ * 180}˝)

for success in file_validate_success:
print(f˝n- 쿼리 {success[‘query_no‘]}: {success[‘sql_first_line‘]}˝)
print(f˝ 결과 행 수: {success[‘row_count‘]}˝)


# ==================== 전체 결과에 누적 ====================
overall_validate_success.extend(file_validate_success)
overall_validate_fail.extend(file_validate_fail)


# ==================== 최종 결과 요약 ====================
print(f˝n{‘=‘ * 180}˝)
print(˝✅ 전체 조회 쿼리 처리 완료˝)
print(f˝{‘=‘ * 180}˝)

print(˝n📊 [조회 쿼리 최종 결과]˝)
print(f˝처리된 파일 수 : {len(validate_query_files)}˝)
print(f˝✅ 성공 쿼리 수 : {len(overall_validate_success)}˝)
print(f˝❌ 실패 쿼리 수 : {len(overall_validate_fail)}˝)


# ==================== 전체 실패 목록 ====================
if overall_validate_fail:
print(f˝n{‘=‘ * 180}˝)
print(˝❌ 전체 실패 목록˝)
print(f˝{‘=‘ * 180}˝)

for fail in overall_validate_fail:
print(f˝n파일: {fail[‘file‘]}˝)
print(f˝쿼리: {fail[‘query_no‘]}˝)
print(f˝SQL 첫 줄: {fail[‘sql_first_line‘]}˝)
print(f˝오류: {fail[‘error‘]}˝)


# ==================== 전체 성공 목록 ====================
if overall_validate_success:
print(f˝n{‘=‘ * 180}˝)
print(˝✅ 전체 성공 목록˝)
print(f˝{‘=‘ * 180}˝)

for success in overall_validate_success:
print(f˝n파일: {success[‘file‘]}˝)
print(f˝쿼리: {success[‘query_no‘]}˝)
print(f˝SQL 첫 줄: {success[‘sql_first_line‘]}˝)
print(f˝결과 행 수: {success[‘row_count‘]}˝)


print(f˝n[전체종료] {datetime.now(KST).strftime(‘%Y%m%d_%H%M%S‘)}˝)
print(f˝{‘=‘ * 180}˝)