Untitled

        # ============== INCREMENTAL DATE-PARTITIONED WRITE ==============
        if args.step in ['all', 'write']:
            print("\n" + "=" * 80)
            print("[STEP 7/7] Writing to Iceberg table (INCREMENTAL)...")
            print("=" * 80)
            
            target_table = f"iceberg.compo_pivot_database.{args.target_table}"
            print(f"[STEP 7/7] Target table: {target_table}")
            
            # Date partition analizi
            date_partitions = final_df.select("CONFIRM_DATE").distinct().collect()
            partition_dates = sorted([row.CONFIRM_DATE for row in date_partitions])
            
            print(f"[STEP 7/7] Incremental load strategy:")
            print(f"  - Processing dates: {len(partition_dates)} days")
            print(f"  - Date range: {partition_dates[0]} to {partition_dates[-1]}")
            
            # Check if table exists
            try:
                existing_count = spark.sql(f"SELECT COUNT(*) FROM {target_table}").collect()[0][0]
                table_exists = True
                print(f"[STEP 7/7] ✓ Table exists with {existing_count:,} rows")
                
                # Eski partition'ları sil (overwrite mode için)
                if args.write_mode == 'overwrite':
                    print(f"[STEP 7/7] Deleting existing partitions for date range...")
                    for date_val in partition_dates:
                        try:
                            spark.sql(f"""
                                DELETE FROM {target_table} 
                                WHERE CONFIRM_DATE = '{date_val}'
                            """)
                            print(f"[STEP 7/7]   ✓ Deleted partition: {date_val}")
                        except Exception as e:
                            print(f"[STEP 7/7]   ⚠ Could not delete {date_val}: {e}")
                    
                    write_mode = "append"  # Silme sonrası append kullan
                else:
                    write_mode = args.write_mode
                    
            except Exception as e:
                table_exists = False
                write_mode = "overwrite"  # İlk kez oluşturulacak
                print(f"[STEP 7/7] Table doesn't exist, will create new")
            
            # Repartition by date
            print(f"[STEP 7/7] Repartitioning by CONFIRM_DATE...")
            final_ordered = final_df.repartition("CONFIRM_DATE")
            
            print(f"[STEP 7/7] Starting incremental write (mode: {write_mode})...")
            write_start = datetime.now()
            
            # INCREMENTAL PARTITIONED WRITE
            (final_ordered.write
             .format("iceberg")
             .mode(write_mode)
             .partitionBy("CONFIRM_DATE")  # 🔥 Date partition
             .option("write.parquet.compression-codec", "snappy")
             .option("write.target-file-size-bytes", "536870912")
             .option("write.distribution-mode", "hash")
             .option("write.spark.fanout.enabled", "true")
             .option("write.wap.enabled", "false")
             .saveAsTable(target_table))
            
            write_end = datetime.now()
            write_duration = (write_end - write_start).total_seconds()
            
            print(f"[STEP 7/7] ✓ Incremental write completed!")
            print(f"[STEP 7/7] Duration: {write_duration:.1f}s ({write_duration/60:.1f} min)")
            print(f"[STEP 7/7] Throughput: {final_rows/write_duration:.0f} rows/sec")
            print(f"[STEP 7/7] Partitions written: {len(partition_dates)}")
            
            # Validation
            validation = spark.sql(f"SELECT COUNT(*) as cnt FROM {target_table}")
            written_count = validation.collect()[0]['cnt']
            print(f"[STEP 7/7] ✓ Total rows in table: {written_count:,}")
            print(f"[STEP 7/7] ✓ New rows added: {final_rows:,}")