Untitled
alp
plain_text
9 months ago
17 kB
37
Indexable
"""
Iceberg Tablosu Optimizasyon Scripti
- Mevcut düz tabloyu aylık partition'lara dönüştürme
- Bloom filter ekleme
- Z-ordering ile optimize etme
- Sorgu performansını artırma
"""
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from datetime import datetime
import time
# ============================================================================
# KONFIGURASYON
# ============================================================================
# Tablo bilgileri
SOURCE_TABLE = "iceberg.sompodatabase2.T001PSRCVP"
TARGET_TABLE = "iceberg.sompodatabase2.T001PSRCVP_partitioned"
TEMP_TABLE = "iceberg.sompodatabase2.T001PSRCVP_temp"
# Kolon tanımları
DATE_COL = "CONFIRM_DATE"
PK_COLS = ["FIRM_CODE", "COMPANY_CODE", "PRODUCT_NO", "POLICY_NO", "RENEWAL_NO", "ENDORS_NO"]
# Optimize parametreleri
TARGET_FILE_SIZE = 512 * 1024 * 1024 # 512 MB
MAX_CONCURRENT_REWRITES = 8
ENABLE_BLOOM_FILTER = True
BLOOM_FILTER_FPP = 0.05
# ============================================================================
# YARDIMCI FONKSİYONLAR
# ============================================================================
def log(message):
"""Zaman damgalı log mesajı"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] {message}")
def format_bytes(bytes_val):
"""Byte'ı okunabilir formata çevir"""
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if bytes_val < 1024.0:
return f"{bytes_val:.2f} {unit}"
bytes_val /= 1024.0
return f"{bytes_val:.2f} PB"
def get_table_stats(spark, table_name):
"""Tablo istatistiklerini getir"""
try:
stats = spark.sql(f"""
SELECT
COUNT(*) as file_count,
SUM(file_size_in_bytes) as total_size,
AVG(file_size_in_bytes) as avg_file_size,
MIN(file_size_in_bytes) as min_file_size,
MAX(file_size_in_bytes) as max_file_size
FROM {table_name}.files
""").first()
return stats
except:
return None
# ============================================================================
# 1. MEVCUT TABLO DURUMUNU KONTROL ET
# ============================================================================
def check_current_state(spark, table_name, date_col):
"""Mevcut tablo durumunu kontrol et"""
log("="*80)
log("ADIM 1: Mevcut Tablo Durumu Kontrol Ediliyor")
log("="*80)
# Dosya istatistiklerinden bilgi al (daha hızlı)
stats = get_table_stats(spark, table_name)
if stats:
log(f"Dosya sayısı: {stats['file_count']}")
log(f"Toplam boyut: {format_bytes(stats['total_size'])}")
log(f"Ortalama dosya boyutu: {format_bytes(stats['avg_file_size'])}")
log(f"Min dosya boyutu: {format_bytes(stats['min_file_size'])}")
log(f"Max dosya boyutu: {format_bytes(stats['max_file_size'])}")
# Sadece MIN/MAX al (COUNT DISTINCT yapma - çok yavaş)
date_range = spark.sql(f"""
SELECT
MIN({date_col}) as min_date,
MAX({date_col}) as max_date
FROM {table_name}
""").first()
min_date = date_range['min_date']
max_date = date_range['max_date']
# Ay sayısını hesapla
if min_date and max_date:
month_count = (max_date.year - min_date.year) * 12 + (max_date.month - min_date.month) + 1
else:
month_count = 0
log(f"Tarih aralığı: {min_date} - {max_date}")
log(f"Tahmini ay sayısı: {month_count}")
# Toplam satır sayısını table properties'den al (varsa)
try:
row_count_estimate = spark.sql(f"SELECT COUNT(*) as cnt FROM {table_name}.files").first()['cnt']
log(f"Dosya metadata'sından tahmini satır bilgisi alındı")
except:
log(f"NOT: Tam satır sayımı yapılmadı (çok uzun sürer)")
row_count_estimate = None
return row_count_estimate, {'min_date': min_date, 'max_date': max_date, 'month_count': month_count}
# ============================================================================
# 2. PARTITION'LI YENİ TABLO OLUŞTUR
# ============================================================================
def create_partitioned_table(spark, source_table, target_table, date_col, pk_cols):
"""Aylık partition'lı yeni tablo oluştur"""
log("="*80)
log("ADIM 2: Partition'lı Yeni Tablo Oluşturuluyor")
log("="*80)
# Eski tabloyu sil (eğer varsa)
spark.sql(f"DROP TABLE IF EXISTS {target_table}")
log(f"Eski tablo silindi (eğer varsa): {target_table}")
# Tablo özelliklerini al
source_schema = spark.table(source_table).schema
# Yeni partition'lı tablo oluştur
spark.sql(f"""
CREATE TABLE {target_table}
USING iceberg
PARTITIONED BY (months({date_col}))
TBLPROPERTIES (
'write.format.default' = 'parquet',
'write.parquet.compression-codec' = 'zstd',
'write.metadata.compression-codec' = 'gzip',
'write.target-file-size-bytes' = '{TARGET_FILE_SIZE}',
'write.distribution-mode' = 'hash',
'commit.retry.num-retries' = '10',
'commit.retry.min-wait-ms' = '100',
'history.expire.max-snapshot-age-ms' = '432000000'
)
AS SELECT * FROM {source_table} WHERE 1=0
""")
log(f"Yeni partition'lı tablo oluşturuldu: {target_table}")
# ============================================================================
# 3. VERİYİ AY AY KOPYALA
# ============================================================================
def migrate_data_by_month(spark, source_table, target_table, date_col):
"""Veriyi ay ay yeni tabloya kopyala"""
log("="*80)
log("ADIM 3: Veri Ay Ay Kopyalanıyor")
log("="*80)
# Tüm ayları listele
months = spark.sql(f"""
SELECT DISTINCT DATE_TRUNC('MONTH', {date_col}) as month
FROM {source_table}
ORDER BY month
""").collect()
total_months = len(months)
log(f"Toplam {total_months} ay kopyalanacak")
# Her ayı ayrı ayrı kopyala
for idx, row in enumerate(months, 1):
month = row['month']
month_str = month.strftime('%Y-%m')
log(f"\n[{idx}/{total_months}] {month_str} ayı işleniyor...")
start_time = time.time()
# Veriyi kopyala
spark.sql(f"""
INSERT INTO {target_table}
SELECT * FROM {source_table}
WHERE DATE_TRUNC('MONTH', {date_col}) = DATE'{month}'
""")
elapsed = time.time() - start_time
# Kopyalanan satır sayısını kontrol et
count = spark.sql(f"""
SELECT COUNT(*) as cnt FROM {target_table}
WHERE DATE_TRUNC('MONTH', {date_col}) = DATE'{month}'
""").first()['cnt']
log(f"✓ {month_str}: {count:,} satır kopyalandı ({elapsed:.1f} saniye)")
log("\n✓ Tüm veri başarıyla kopyalandı!")
# ============================================================================
# 4. BLOOM FILTER EKLE
# ============================================================================
def add_bloom_filters(spark, table_name, pk_cols, fpp):
"""Bloom filter ekle"""
log("="*80)
log("ADIM 4: Bloom Filter Ekleniyor")
log("="*80)
cols_str = ",".join(pk_cols)
spark.sql(f"""
ALTER TABLE {table_name}
SET TBLPROPERTIES (
'bloom_filter.columns' = '{cols_str}',
'bloom_filter.fpp' = '{fpp}'
)
""")
log(f"✓ Bloom filter eklendi: {cols_str}")
log(f"✓ False Positive Probability: {fpp}")
# ============================================================================
# 5. HER PARTITION'I OPTİMİZE ET
# ============================================================================
def optimize_partitions(spark, table_name, date_col, pk_cols):
"""Her partition'ı ayrı ayrı optimize et"""
log("="*80)
log("ADIM 5: Partition'lar Optimize Ediliyor")
log("="*80)
# Tüm partition'ları listele
partitions = spark.sql(f"""
SELECT DISTINCT DATE_TRUNC('MONTH', {date_col}) as month
FROM {table_name}
ORDER BY month DESC
""").collect()
total_partitions = len(partitions)
log(f"Toplam {total_partitions} partition optimize edilecek")
sort_order = ",".join(pk_cols)
# Her partition'ı optimize et
for idx, row in enumerate(partitions, 1):
month = row['month']
month_str = month.strftime('%Y-%m')
log(f"\n[{idx}/{total_partitions}] {month_str} optimize ediliyor...")
start_time = time.time()
try:
# Rewrite data files with sort
result = spark.sql(f"""
CALL iceberg.system.rewrite_data_files(
table => '{table_name}',
strategy => 'sort',
sort_order => '{sort_order}',
where => "{date_col} >= DATE'{month}' AND {date_col} < DATE'{month}' + INTERVAL 1 MONTH",
options => map(
'target-file-size-bytes', '{TARGET_FILE_SIZE}',
'partial-progress.enabled', 'true',
'partial-progress.max-commits', '20',
'max-concurrent-file-group-rewrites', '{MAX_CONCURRENT_REWRITES}'
)
)
""").first()
elapsed = time.time() - start_time
log(f"✓ {month_str} optimize edildi:")
log(f" - Yeniden yazılan dosya: {result['rewritten_data_files_count']}")
log(f" - Eklenen dosya: {result['added_data_files_count']}")
log(f" - Yeniden yazılan veri: {format_bytes(result['rewritten_bytes_count'])}")
log(f" - Süre: {elapsed:.1f} saniye")
except Exception as e:
log(f"✗ HATA: {month_str} optimize edilemedi: {str(e)}")
continue
log("\n✓ Tüm partition'lar optimize edildi!")
# ============================================================================
# 6. MANIFEST DOSYALARINI OPTİMİZE ET
# ============================================================================
def optimize_manifests(spark, table_name):
"""Manifest dosyalarını optimize et"""
log("="*80)
log("ADIM 6: Manifest Dosyaları Optimize Ediliyor")
log("="*80)
start_time = time.time()
result = spark.sql(f"""
CALL iceberg.system.rewrite_manifests(
table => '{table_name}'
)
""").first()
elapsed = time.time() - start_time
log(f"✓ Manifest optimize edildi:")
log(f" - Yeniden yazılan manifest: {result['rewritten_manifests_count']}")
log(f" - Eklenen manifest: {result['added_manifests_count']}")
log(f" - Süre: {elapsed:.1f} saniye")
# ============================================================================
# 7. ESKİ SNAPSHOT'LARI TEMİZLE
# ============================================================================
def cleanup_old_snapshots(spark, table_name, retain_days=7):
"""Eski snapshot'ları temizle"""
log("="*80)
log("ADIM 7: Eski Snapshot'lar Temizleniyor")
log("="*80)
# Expire snapshots
spark.sql(f"""
CALL iceberg.system.expire_snapshots(
table => '{table_name}',
older_than => DATE_SUB(CURRENT_DATE(), {retain_days}),
retain_last => 5
)
""")
log(f"✓ {retain_days} günden eski snapshot'lar temizlendi")
# Remove orphan files
spark.sql(f"""
CALL iceberg.system.remove_orphan_files(
table => '{table_name}',
older_than => TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {retain_days} DAY)
)
""")
log(f"✓ Orphan dosyalar temizlendi")
# ============================================================================
# 8. ESKİ TABLOYU YENİSİYLE DEĞİŞTİR
# ============================================================================
def swap_tables(spark, source_table, target_table, temp_table):
"""Eski tabloyu yenisiyle değiştir"""
log("="*80)
log("ADIM 8: Tablolar Değiştiriliyor")
log("="*80)
# Eski tabloyu temp'e taşı
spark.sql(f"ALTER TABLE {source_table} RENAME TO {temp_table}")
log(f"✓ Eski tablo temp'e taşındı: {temp_table}")
# Yeni tabloyu eski adla değiştir
spark.sql(f"ALTER TABLE {target_table} RENAME TO {source_table}")
log(f"✓ Yeni tablo eski adla değiştirildi: {source_table}")
log("\n" + "!"*80)
log("! DİKKAT: Eski tablo şimdi temp olarak saklanıyor: {temp_table}")
log("! Her şey yolundaysa şu komutla silebilirsiniz:")
log(f"! spark.sql('DROP TABLE {temp_table}')")
log("!"*80)
# ============================================================================
# 9. FİNAL İSTATİSTİKLER
# ============================================================================
def show_final_stats(spark, table_name, date_col):
"""Final istatistiklerini göster"""
log("="*80)
log("FİNAL İSTATİSTİKLER")
log("="*80)
# Genel istatistikler
stats = get_table_stats(spark, table_name)
if stats:
log(f"\nGenel İstatistikler:")
log(f" - Dosya sayısı: {stats['file_count']}")
log(f" - Toplam boyut: {format_bytes(stats['total_size'])}")
log(f" - Ortalama dosya boyutu: {format_bytes(stats['avg_file_size'])}")
# Partition bazında istatistikler
log(f"\nPartition Bazında İstatistikler:")
partition_stats = spark.sql(f"""
SELECT
DATE_TRUNC('MONTH', {date_col}) as month,
COUNT(*) as file_count,
ROUND(SUM(file_size_in_bytes) / 1024 / 1024 / 1024, 2) as size_gb,
ROUND(AVG(file_size_in_bytes) / 1024 / 1024, 2) as avg_file_mb
FROM {table_name}.files
GROUP BY DATE_TRUNC('MONTH', {date_col})
ORDER BY month DESC
LIMIT 10
""")
partition_stats.show(truncate=False)
# Performans önerileri
log("\n" + "="*80)
log("PERFORMANS ÖNERİLERİ")
log("="*80)
log("""
1. Sorgularınızda mutlaka partition filtresi kullanın:
WHERE CONFIRM_DATE >= '2024-01-01' AND CONFIRM_DATE < '2024-02-01'
2. Primary key kolonlarında filtre kullanın (Bloom filter aktif):
WHERE FIRM_CODE = 'XXX' AND COMPANY_CODE = 'YYY'
3. Broadcast join için küçük tablolar:
df.join(broadcast(small_df), "key")
4. Predicate pushdown için Parquet filtreleri:
spark.sql("SELECT * FROM table WHERE col > 100")
5. Düzenli bakım yapın:
- Ayda bir: expire_snapshots + remove_orphan_files
- 3 ayda bir: rewrite_manifests
- Yeni veri geldiğinde: ilgili partition'ı optimize edin
""")
# ============================================================================
# ANA FONKSİYON
# ============================================================================
def main():
"""Ana optimizasyon fonksiyonu"""
# Spark session
spark = SparkSession.builder.getOrCreate()
overall_start = time.time()
try:
# 1. Mevcut durumu kontrol et
row_count, date_range = check_current_state(spark, SOURCE_TABLE, DATE_COL)
# 2. Partition'lı yeni tablo oluştur
create_partitioned_table(spark, SOURCE_TABLE, TARGET_TABLE, DATE_COL, PK_COLS)
# 3. Veriyi ay ay kopyala
migrate_data_by_month(spark, SOURCE_TABLE, TARGET_TABLE, DATE_COL)
# 4. Bloom filter ekle
if ENABLE_BLOOM_FILTER:
add_bloom_filters(spark, TARGET_TABLE, PK_COLS, BLOOM_FILTER_FPP)
# 5. Partition'ları optimize et
optimize_partitions(spark, TARGET_TABLE, DATE_COL, PK_COLS)
# 6. Manifest'leri optimize et
optimize_manifests(spark, TARGET_TABLE)
# 7. Eski snapshot'ları temizle
cleanup_old_snapshots(spark, TARGET_TABLE)
# 8. Tabloları değiştir
swap_tables(spark, SOURCE_TABLE, TARGET_TABLE, TEMP_TABLE)
# 9. Final istatistikler
show_final_stats(spark, SOURCE_TABLE, DATE_COL)
overall_elapsed = time.time() - overall_start
log(f"\n{'='*80}")
log(f"✓ TÜM İŞLEMLER BAŞARIYLA TAMAMLANDI!")
log(f"✓ Toplam süre: {overall_elapsed/60:.1f} dakika")
log(f"{'='*80}")
except Exception as e:
log(f"\n{'='*80}")
log(f"✗ HATA OLUŞTU: {str(e)}")
log(f"{'='*80}")
raise
# ============================================================================
# ÇALIŞTIR
# ============================================================================
if __name__ == "__main__":
main()Editor is loading...
Leave a Comment