Untitled
alp
plain_text
9 months ago
8.8 kB
15
Indexable
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
from datetime import datetime, timedelta
from pyspark.sql import SparkSession, functions as F
# ---------- MSSQL JDBC ----------
jdbc_url = "jdbc:sqlserver://10.81.24.55:1433;databaseName=WINSURE;encrypt=true;trustServerCertificate=true"
jdbc_props = {
"user": "bentego",
"password":
"driver": "com.microsoft.sqlserver.jdbc.SQLServerDriver",
"fetchsize": "50000",
"queryTimeout": "0",
"responseBuffering": "adaptive",
"packetSize": "32768",
"sendStringParametersAsUnicode": "false",
}
# ---------- Sabitler ----------
PK_COLS = ["FIRM_CODE", "COMPANY_CODE", "PRODUCT_NO", "POLICY_NO", "RENEWAL_NO", "ENDORS_NO"]
DATE_COL = "CONFIRM_DATE"
CATALOG = "iceberg"
TARGET_DB = "somcodatabase2"
def parse_date(param, full_tbl, spark):
if not param or param.lower() == "auto":
try:
if spark.catalog.tableExists(full_tbl):
r = spark.sql(f"SELECT MAX({DATE_COL}) AS d FROM {full_tbl}").collect()[0]["d"]
if r:
return r.strftime("%Y-%m-%d") + " 00:00:00.000"
except Exception:
pass
return "1900-01-01 00:00:00.000"
p = param.lower()
now = datetime.now()
if p.startswith("last_") and p.endswith("h"):
return (now - timedelta(hours=int(p[5:-1]))).strftime("%Y-%m-%d %H:%M:%S.000")
if p.startswith("last_") and p.endswith("d"):
return (now - timedelta(days=int(p[5:-1]))).strftime("%Y-%m-%d %H:%M:%S.000")
return param
def values_table_from_batch(batch_pks_df):
"""(VALUES (...), ...) AS p(...) türetilmiş tablosu döner (CTE YOK)."""
rows = batch_pks_df.collect() # batch ~100
if not rows:
return None
vals = []
for r in rows:
vals.append(
f"({int(r.FIRM_CODE)},{int(r.COMPANY_CODE)},{int(r.PRODUCT_NO)},"
f"{int(r.POLICY_NO)},{int(r.RENEWAL_NO)},{int(r.ENDORS_NO)})"
)
values_str = ",\n ".join(vals)
return (
f"(VALUES\n {values_str}\n ) AS p("
"FIRM_CODE, COMPANY_CODE, PRODUCT_NO, POLICY_NO, RENEWAL_NO, ENDORS_NO)"
)
def microbatch_update(spark, table, start_date):
full_tbl = f"{CATALOG}.{TARGET_DB}.{table}"
print(f"\n[{table}] Start | from: {start_date}")
# 1) Kaç benzersiz PK değişmiş?
count_query = f"""
SELECT COUNT(DISTINCT
CAST(FIRM_CODE AS VARCHAR)+'|'+CAST(COMPANY_CODE AS VARCHAR)+'|'+CAST(PRODUCT_NO AS VARCHAR)+'|'+
CAST(POLICY_NO AS VARCHAR)+'|'+CAST(RENEWAL_NO AS VARCHAR)+'|'+CAST(ENDORS_NO AS VARCHAR)
) AS cnt
FROM [WINSURE].[ETLUSR].[T001POLMAS_KEY_LOG] WITH (NOLOCK)
WHERE [DATE] > '{start_date}'
AND FIRM_CODE IN (2,3)
AND COMPANY_CODE IN (2,3)
"""
pk_count = (
spark.read.format("jdbc")
.option("url", jdbc_url)
.option("query", count_query)
.options(**jdbc_props)
.load()
.collect()[0]["cnt"]
)
if pk_count == 0:
print(f"[{table}] No changes.\n")
return
print(f"[{table}] Changed PKs: {pk_count}")
batch_size = 100
num_batches = (pk_count // batch_size) + 1
total_rows = 0
for i in range(num_batches):
offset = i * batch_size
# 2) CTE KULLANMADAN batch PK'ları (OFFSET/FETCH)
pk_query = f"""
SELECT DISTINCT
FIRM_CODE, COMPANY_CODE, PRODUCT_NO,
POLICY_NO, RENEWAL_NO, ENDORS_NO
FROM [WINSURE].[ETLUSR].[T001POLMAS_KEY_LOG] WITH (NOLOCK)
WHERE [DATE] > '{start_date}'
AND FIRM_CODE IN (2,3)
AND COMPANY_CODE IN (2,3)
ORDER BY FIRM_CODE, COMPANY_CODE, PRODUCT_NO, POLICY_NO, RENEWAL_NO, ENDORS_NO
OFFSET {offset} ROWS
FETCH NEXT {batch_size} ROWS ONLY
"""
batch_pks = (
spark.read.format("jdbc")
.option("url", jdbc_url)
.option("query", pk_query)
.options(**jdbc_props)
.load()
)
if batch_pks.rdd.isEmpty():
continue
# 3) Iceberg DELETE (idempotent)
batch_pks.createOrReplaceTempView("batch_pks")
if spark.catalog.tableExists(full_tbl):
cond = " AND ".join([f"{full_tbl}.{c} = p.{c}" for c in PK_COLS])
spark.sql(f"DELETE FROM {full_tbl} WHERE EXISTS (SELECT 1 FROM batch_pks p WHERE {cond})")
# 4) MSSQL'den JOIN ile veri çek (OR yok, CTE yok)
values_tbl = values_table_from_batch(batch_pks)
if not values_tbl:
spark.catalog.dropTempView("batch_pks")
continue
if table == "T001POLMAS":
data_query = f"""
SELECT t.*
FROM fiba.T001POLMAS t WITH (NOLOCK, FORCESEEK)
JOIN {values_tbl}
ON t.FIRM_CODE = p.FIRM_CODE
AND t.COMPANY_CODE = p.COMPANY_CODE
AND t.PRODUCT_NO = p.PRODUCT_NO
AND t.POLICY_NO = p.POLICY_NO
AND t.RENEWAL_NO = p.RENEWAL_NO
AND t.ENDORS_NO = p.ENDORS_NO
"""
else:
data_query = f"""
SELECT t.*, CAST(pm.{DATE_COL} AS DATE) AS {DATE_COL}_CAST
FROM fiba.{table} t WITH (NOLOCK, FORCESEEK)
JOIN {values_tbl}
ON t.FIRM_CODE = p.FIRM_CODE
AND t.COMPANY_CODE = p.COMPANY_CODE
AND t.PRODUCT_NO = p.PRODUCT_NO
AND t.POLICY_NO = p.POLICY_NO
AND t.RENEWAL_NO = p.RENEWAL_NO
AND t.ENDORS_NO = p.ENDORS_NO
JOIN fiba.T001POLMAS pm WITH (NOLOCK, FORCESEEK)
ON pm.FIRM_CODE = p.FIRM_CODE
AND pm.COMPANY_CODE = p.COMPANY_CODE
AND pm.PRODUCT_NO = p.PRODUCT_NO
AND pm.POLICY_NO = p.POLICY_NO
AND pm.RENEWAL_NO = p.RENEWAL_NO
AND pm.ENDORS_NO = p.ENDORS_NO
"""
new_data = (
spark.read.format("jdbc")
.option("url", jdbc_url)
.option("query", data_query)
.options(**jdbc_props)
.load()
)
# 5) POLMAS dışı tablolar: _CAST → CONFIRM_DATE
cast_col = f"{DATE_COL}_CAST"
if cast_col in new_data.columns:
if DATE_COL in new_data.columns:
new_data = new_data.drop(DATE_COL)
new_data = new_data.withColumnRenamed(cast_col, DATE_COL)
# 6) pk_hash ekle
new_data = new_data.withColumn("pk_hash", F.xxhash64(*[F.col(c) for c in PK_COLS]))
# 7) Hedef tablo yoksa oluştur
if not spark.catalog.tableExists(full_tbl):
(
new_data.limit(0)
.write.format("iceberg")
.tableProperty("write.format.default", "parquet")
.tableProperty("write.parquet.compression-codec", "zstd")
.saveAsTable(full_tbl)
)
spark.sql(f"ALTER TABLE {full_tbl} ADD PARTITION FIELD months({DATE_COL})")
# 8) Şema hizalama
target_cols = [f.name for f in spark.table(full_tbl).schema]
new_data = new_data.select(
*[F.col(c) if c in new_data.columns else F.lit(None).alias(c) for c in target_cols]
)
# 9) Yaz (append)
(
new_data.coalesce(2)
.write.format("iceberg")
.mode("append")
.saveAsTable(full_tbl)
)
total_rows += new_data.count()
print(f" Batch {i+1}/{num_batches} ✓")
spark.catalog.dropTempView("batch_pks")
spark.catalog.clearCache()
print(f"[{table}] Done | total rows appended: {total_rows:,}\n")
def main():
if len(sys.argv) < 2:
print("Usage: spark-submit update_script.py <table> [date]")
sys.exit(1)
table = sys.argv[1]
date_param = sys.argv[2] if len(sys.argv) > 2 else None
spark = SparkSession.builder.appName(f"Update-{table}").getOrCreate()
try:
full_tbl = f"{CATALOG}.{TARGET_DB}.{table}"
start_date = parse_date(date_param, full_tbl, spark)
print("=" * 60)
print(f"MICRO-BATCH UPDATE: {table}")
print("=" * 60)
microbatch_update(spark, table, start_date)
print("=" * 60)
print(f"✔ {table} DONE")
print("=" * 60)
except Exception as e:
import traceback
print(f"\n✘ ERROR: {e}\n")
traceback.print_exc()
sys.exit(1)
finally:
spark.stop()
if __name__ == "__main__":
main()
Editor is loading...
Leave a Comment