from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HudiCOWWrite") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog") \
    .getOrCreate()

# Sample data
data = [(1, "Alice", "Engineering", 1000),
        (2, "Bob",   "Marketing",   2000),
        (3, "Carol", "Engineering", 3000)]
df = spark.createDataFrame(data, ["emp_id", "name", "dept", "ts"])

# Hudi COW options
hudi_options = {
    "hoodie.table.name": "employees",
    "hoodie.datasource.write.table.type": "COPY_ON_WRITE",  # ← COW
    "hoodie.datasource.write.recordkey.field": "emp_id",       # ← primary key
    "hoodie.datasource.write.precombine.field": "ts",            # ← latest wins
    "hoodie.datasource.write.partitionpath.field": "dept",       # ← partition
    "hoodie.datasource.write.hive_style_partitioning": "true",
    "hoodie.upsert.shuffle.parallelism": "2",
    "hoodie.insert.shuffle.parallelism": "2",
}

table_path = "s3://my-bucket/hudi/employees"

# Write (UPSERT is the standard operation)
df.write.format("hudi") \
    .options(**hudi_options) \
    .option("hoodie.datasource.write.operation", "upsert") \
    .mode("append") \               # always use 'append' with Hudi
    .save(table_path)

print("COW table written successfully!")

# MOR write — only change is the table.type option
hudi_mor_options = {
    "hoodie.table.name": "employees_mor",
    "hoodie.datasource.write.table.type": "MERGE_ON_READ",  # ← MOR
    "hoodie.datasource.write.recordkey.field": "emp_id",
    "hoodie.datasource.write.precombine.field": "ts",
    "hoodie.datasource.write.partitionpath.field": "dept",
    "hoodie.datasource.write.hive_style_partitioning": "true",
    # Compaction settings
    "hoodie.compact.inline": "false",                     # async compaction
    "hoodie.compact.inline.max.delta.commits": "5",         # compact after 5 commits
    "hoodie.upsert.shuffle.parallelism": "4",
}

df.write.format("hudi") \
    .options(**hudi_mor_options) \
    .option("hoodie.datasource.write.operation", "upsert") \
    .mode("append") \
    .save("s3://my-bucket/hudi/employees_mor")

hudi_options = {
    # ... other options ...
    "hoodie.index.type": "BLOOM",                       # default
    "hoodie.bloom.index.parallelism": "100",             # lookup parallelism
    "hoodie.bloom.index.filter.type": "DYNAMIC_V0",     # adaptive filter
    "hoodie.bloom.index.keys.per.bucket": "100000",     # records per bloom filter
}

hudi_options = {
    "hoodie.index.type": "SIMPLE",
    "hoodie.simple.index.parallelism": "50",
}

hudi_options = {
    "hoodie.index.type": "HBASE",
    "hoodie.index.hbase.zkquorum": "zk-host:2181",    # ZooKeeper quorum
    "hoodie.index.hbase.zkport": "2181",
    "hoodie.index.hbase.table": "hudi_index_employees",# HBase table for index
    "hoodie.index.hbase.get.batch.size": "100",
}

hudi_options = {
    "hoodie.index.type": "BUCKET",
    "hoodie.storage.layout.type": "BUCKET",
    "hoodie.storage.layout.bucket.num": "128",          # number of buckets per partition
    "hoodie.index.bucket.engine": "SIMPLE",
}

# UPSERT: handles both new and existing records
df.write.format("hudi") \
    .options(**hudi_options) \
    .option("hoodie.datasource.write.operation", "upsert") \
    .mode("append") \
    .save(table_path)

# Use case: CDC stream — process insert + update events together
# Hudi index determines whether each record is new or existing

# INSERT: no index lookup — faster, but no dedup
df.write.format("hudi") \
    .options(**hudi_options) \
    .option("hoodie.datasource.write.operation", "insert") \
    .mode("append") \
    .save(table_path)

# Use case: Append-only event log where each record is guaranteed unique
# E.g., clickstream events with unique event_id

# BULK_INSERT: for initial table population (TBs of data)
hudi_bulk_options = {
    "hoodie.table.name": "orders",
    "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.precombine.field": "updated_at",
    "hoodie.datasource.write.partitionpath.field": "order_date",
    # Bulk insert specific
    "hoodie.bulkinsert.shuffle.parallelism": "200",           # high parallelism
    "hoodie.datasource.write.row.writer.enable": "true",       # row-based writer (faster)
}

historical_df.write.format("hudi") \
    .options(**hudi_bulk_options) \
    .option("hoodie.datasource.write.operation", "bulk_insert") \
    .mode("append") \
    .save(table_path)

# After bulk insert, switch to UPSERT for incremental updates

# Delete specific records by key
from pyspark.sql.functions import lit

# Records to delete — only need the key fields + partition + ts
delete_df = spark.createDataFrame(
    [(2, "Marketing", 9999)],   # emp_id=2, dept=Marketing
    ["emp_id", "dept", "ts"]
)

delete_df.write.format("hudi") \
    .options(**hudi_options) \
    .option("hoodie.datasource.write.operation", "delete") \
    .mode("append") \
    .save(table_path)

# Verify deletion
read_df = spark.read.format("hudi").load(table_path)
read_df.filter("emp_id = 2").show()  # → empty result

# Snapshot query — latest data (includes delta log merges for MOR)
snapshot_df = spark.read \
    .format("hudi") \
    .load("s3://my-bucket/hudi/employees_mor")

snapshot_df.show()

# Hudi adds meta columns to every read:
# _hoodie_commit_time, _hoodie_commit_seqno, _hoodie_record_key,
# _hoodie_partition_path, _hoodie_file_name
snapshot_df.select("_hoodie_commit_time", "emp_id", "name").show()

# Read-Optimized: only base files, no log merge → fastest reads
ro_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "read_optimized") \
    .load("s3://my-bucket/hudi/employees_mor")

ro_df.show()
# ⚠️ Data is only as fresh as the last compaction!
# Records written to delta logs since last compaction won't appear here

# Incremental query: only records changed after a given commit
incremental_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", "20240101120000") \
    .load("s3://my-bucket/hudi/employees")

incremental_df.show()
# Returns only records committed AFTER 20240101120000

# You can also set an end time to read a specific window:
incremental_window_df = spark.read \
    .format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", "20240101120000") \
    .option("hoodie.datasource.read.end.instanttime",   "20240101130000") \
    .load("s3://my-bucket/hudi/employees")

# Read table as it was at a specific commit timestamp
pit_df = spark.read \
    .format("hudi") \
    .option("as.of.instant", "20240101120000") \   # exact commit time
    .load("s3://my-bucket/hudi/employees")

# Find all commits (to pick a timestamp):
spark.read.format("hudi") \
    .load("s3://my-bucket/hudi/employees") \
    .select("_hoodie_commit_time") \
    .distinct().orderBy("_hoodie_commit_time").show()

hudi_mor_inline = {
    "hoodie.table.name": "orders_mor",
    "hoodie.datasource.write.table.type": "MERGE_ON_READ",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.precombine.field": "updated_at",
    "hoodie.datasource.write.partitionpath.field": "order_date",
    # Inline compaction settings
    "hoodie.compact.inline": "true",                  # enable inline compaction
    "hoodie.compact.inline.max.delta.commits": "5",   # compact every 5 commits
}

df.write.format("hudi") \
    .options(**hudi_mor_inline) \
    .option("hoodie.datasource.write.operation", "upsert") \
    .mode("append") \
    .save(table_path)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Step 1: Write job (no inline compaction)
hudi_mor_async = {
    "hoodie.table.name": "orders_mor",
    "hoodie.datasource.write.table.type": "MERGE_ON_READ",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.precombine.field": "updated_at",
    "hoodie.compact.inline": "false",  # no inline — async instead
}

# Step 2: Separate compaction job (run on schedule via Airflow)
from hudi import HoodieSparkUtils

compaction_config = {
    "spark.serializer": "org.apache.spark.serializer.KryoSerializer",
}

# Using HoodieSparkSqlUtils for Spark SQL-based compaction
spark.sql("""
    CALL hudi.run_compaction(
        table => 'orders_mor',
        path => 's3://my-bucket/hudi/orders_mor'
    )
""")

# Or using command-line: HoodieCompactor.java
# spark-submit --class org.apache.hudi.utilities.HoodieCompactor ...

hudi_options_with_cleaner = {
    # ... other options ...
    "hoodie.cleaner.policy": "KEEP_LATEST_COMMITS",     # most common policy
    "hoodie.cleaner.commits.retained": "10",              # keep last 10 commits
    "hoodie.clean.automatic": "true",                    # auto clean after writes
    "hoodie.clean.async": "true",                        # don't block the write
}

# Alternative: KEEP_LATEST_FILE_VERSIONS
# "hoodie.cleaner.policy": "KEEP_LATEST_FILE_VERSIONS"
# "hoodie.cleaner.fileversions.retained": "3"

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("HudiIntegration") \
    .config("spark.serializer",
            "org.apache.spark.serializer.KryoSerializer") \       # required
    .config("spark.sql.extensions",
            "org.apache.spark.sql.hudi.HoodieSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog",
            "org.apache.spark.sql.hudi.catalog.HoodieCatalog") \
    .config("spark.kryo.registrator",
            "org.apache.spark.HoodieSparkKryoRegistrar") \      # optional perf
    .getOrCreate()

# If using spark-submit, add jars:
# --packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.14.0

from pyspark.sql import Row
from datetime import datetime

# Create sample DataFrame
orders = [
    Row(order_id="ORD001", customer_id=101, amount=250.0,
        status="pending", order_date="2024-01-01", ts=1704067200),
    Row(order_id="ORD002", customer_id=102, amount=175.5,
        status="completed", order_date="2024-01-01", ts=1704067300),
]
df = spark.createDataFrame(orders)

# Complete Hudi write options
hudi_write_options = {
    # Required
    "hoodie.table.name":                              "orders",
    "hoodie.datasource.write.table.type":             "COPY_ON_WRITE",
    "hoodie.datasource.write.recordkey.field":        "order_id",
    "hoodie.datasource.write.precombine.field":       "ts",
    "hoodie.datasource.write.partitionpath.field":    "order_date",
    "hoodie.datasource.write.hive_style_partitioning":"true",
    # Performance
    "hoodie.upsert.shuffle.parallelism":              "4",
    "hoodie.insert.shuffle.parallelism":              "4",
    # Index
    "hoodie.index.type":                              "BLOOM",
    # Cleaner
    "hoodie.cleaner.commits.retained":               "10",
    "hoodie.clean.automatic":                         "true",
}

table_path = "s3://my-bucket/hudi/orders"

# Initial load
df.write.format("hudi") \
    .options(**hudi_write_options) \
    .option("hoodie.datasource.write.operation", "bulk_insert") \
    .mode("append") \
    .save(table_path)

# Incremental update (simulate order status change)
updates = [Row(order_id="ORD001", customer_id=101, amount=250.0,
               status="completed", order_date="2024-01-01", ts=1704070000)]
update_df = spark.createDataFrame(updates)

update_df.write.format("hudi") \
    .options(**hudi_write_options) \
    .option("hoodie.datasource.write.operation", "upsert") \
    .mode("append") \
    .save(table_path)

# Read back — ORD001 should now be "completed"
spark.read.format("hudi").load(table_path).show()

-- Create a COW table with Spark SQL
CREATE TABLE orders (
    order_id     STRING,
    customer_id  INT,
    amount       DOUBLE,
    status       STRING,
    order_date   STRING,
    ts           BIGINT
) USING hudi
LOCATION 's3://my-bucket/hudi/orders_sql'
TBLPROPERTIES (
    type = 'cow',
    primaryKey = 'order_id',
    preCombineField = 'ts'
)
PARTITIONED BY (order_date);

-- Insert data
INSERT INTO orders VALUES
    ('ORD001', 101, 250.0, 'pending',   '2024-01-01', 1704067200),
    ('ORD002', 102, 175.5, 'completed', '2024-01-01', 1704067300);

-- Update via MERGE INTO
MERGE INTO orders t
USING (SELECT 'ORD001' order_id, 'completed' status, 1704070000 ts) s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET t.status = s.status, t.ts = s.ts
WHEN NOT MATCHED THEN INSERT *;

-- Delete a record
DELETE FROM orders WHERE order_id = 'ORD002';

-- Query with time travel
SELECT * FROM orders TIMESTAMP AS OF '2024-01-01 12:00:00';

def get_last_checkpoint(checkpoint_path):
    """Read the last processed commit time from checkpoint"""
    try:
        checkpoint_df = spark.read.format("text").load(checkpoint_path)
        return checkpoint_df.first()[0]
    except:
        return "000"  # default: read from beginning

def save_checkpoint(commit_time, checkpoint_path):
    """Save the latest processed commit time"""
    df = spark.createDataFrame([(commit_time,)], ["commit_time"])
    df.coalesce(1).write.mode("overwrite").format("text").save(checkpoint_path)

# Pipeline run
table_path = "s3://my-bucket/hudi/orders"
checkpoint_path = "s3://my-bucket/checkpoints/orders_pipeline"
silver_path = "s3://my-bucket/silver/orders"

last_commit = get_last_checkpoint(checkpoint_path)
print(f"Processing incremental changes since: {last_commit}")

# Incremental read
incremental_df = spark.read.format("hudi") \
    .option("hoodie.datasource.query.type", "incremental") \
    .option("hoodie.datasource.read.begin.instanttime", last_commit) \
    .load(table_path)

# Process the changes
processed_df = incremental_df \
    .filter("status = 'completed'") \
    .select("order_id", "customer_id", "amount", "order_date",
            "_hoodie_commit_time")

# Write to silver
processed_df.write.mode("append").parquet(silver_path)

# Update checkpoint with latest commit
latest_commit = incremental_df.select("_hoodie_commit_time") \
                               .orderBy("_hoodie_commit_time".desc()).first()[0]
save_checkpoint(latest_commit, checkpoint_path)
print(f"Processed {incremental_df.count()} records. New checkpoint: {latest_commit}")

// EMR Cluster configuration (spark-defaults classification)
[
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.serializer": "org.apache.spark.serializer.KryoSerializer",
      "spark.sql.extensions": "org.apache.spark.sql.hudi.HoodieSparkSessionExtension",
      "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.hudi.catalog.HoodieCatalog"
    }
  }
]

// Add Hudi app to EMR applications list:
// "Applications": [{"Name": "Spark"}, {"Name": "Hudi"}]

hudi_s3_options = {
    "hoodie.table.name": "orders",
    "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.precombine.field": "ts",
    "hoodie.datasource.write.partitionpath.field": "order_date",
    # S3-specific settings
    "hoodie.datasource.write.keygenerator.class":
        "org.apache.hudi.keygen.ComplexKeyGenerator",
    "hoodie.embed.timeline.server": "true",                 # faster on EMR
    "hoodie.embed.timeline.server.port": "26754",
    "hoodie.filesystem.view.type": "EMBEDDED_KV_STORE",     # avoids S3 list calls
    "hoodie.upsert.shuffle.parallelism": "20",
}

# Write to S3 with SSE-S3 encryption
df.write.format("hudi") \
    .options(**hudi_s3_options) \
    .option("hoodie.datasource.write.operation", "upsert") \
    .mode("append") \
    .save("s3://my-data-lake/hudi/orders/")

hudi_glue_sync_options = {
    "hoodie.table.name": "orders",
    "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.precombine.field": "ts",
    "hoodie.datasource.write.partitionpath.field": "order_date",
    "hoodie.datasource.write.hive_style_partitioning": "true",
    # Glue sync options
    "hoodie.datasource.hive_sync.enable": "true",            # enable sync
    "hoodie.datasource.hive_sync.mode": "glue",              # use Glue (not Hive)
    "hoodie.datasource.hive_sync.database": "analytics_db",  # Glue DB name
    "hoodie.datasource.hive_sync.table": "orders",           # Glue table name
    "hoodie.datasource.hive_sync.partition_fields": "order_date",
    "hoodie.datasource.hive_sync.partition_extractor_class":
        "org.apache.hudi.hive.MultiPartKeysValueExtractor",
    "hoodie.datasource.hive_sync.use_jdbc": "false",         # don't use JDBC for Glue
    "hoodie.datasource.hive_sync.region": "us-east-1",
}

df.write.format("hudi") \
    .options(**hudi_glue_sync_options) \
    .option("hoodie.datasource.write.operation", "upsert") \
    .mode("append") \
    .save("s3://my-data-lake/hudi/orders/")

# After this, Athena can query:
# SELECT * FROM analytics_db.orders WHERE order_date = '2024-01-01';

-- COW table: query directly
SELECT order_id, status, amount
FROM analytics_db.orders
WHERE order_date = '2024-01-01'
  AND status = 'completed';

-- MOR table: two views auto-created by Hudi Glue sync
-- Read-Optimized (base files only, fastest):
SELECT * FROM analytics_db.orders_mor_ro LIMIT 100;

-- Real-Time Snapshot (includes log merges, freshest data):
SELECT * FROM analytics_db.orders_mor_rt LIMIT 100;

# Submit Hudi job to EMR Serverless
aws emr-serverless start-job-run \
  --application-id <your-app-id> \
  --execution-role-arn arn:aws:iam::123456789:role/EMRServerlessRole \
  --job-driver '{
    "sparkSubmit": {
      "entryPoint": "s3://my-bucket/scripts/hudi_upsert.py",
      "sparkSubmitParameters": "--conf spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension --jars s3://my-bucket/jars/hudi-spark3.3-bundle_2.12-0.14.0.jar"
    }
  }'