📦 Event Hub Capture to Storage¶

# Read Avro files
df = spark.read.format("avro").load(
    "abfss://eventhub-capture@capturestorage.dfs.core.windows.net/mynamespace/telemetry-events/*/*/*/*/*/*/*.avro"
)

# Extract and parse body
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

# Define schema for event body
schema = StructType([
    StructField("sensor_id", StringType(), True),
    StructField("temperature", DoubleType(), True),
    StructField("timestamp", StringType(), True)
])

# Parse JSON body
parsed_df = df.withColumn(
    "parsed_body",
    from_json(col("Body").cast("string"), schema)
).select(
    "EnqueuedTimeUtc",
    "PartitionId",
    "Offset",
    "SequenceNumber",
    "parsed_body.*"
)

# Show results
parsed_df.show()

# Write to Delta Lake for further processing
parsed_df.write.format("delta").mode("append").save("/mnt/delta/captured_events")

def calculate_optimal_capture_window(
    events_per_second,
    avg_event_size_bytes,
    max_file_size_mb=300,
    min_time_window_sec=60,
    max_time_window_sec=900
):
    """Calculate optimal capture window settings."""
    # Calculate data rate
    bytes_per_second = events_per_second * avg_event_size_bytes
    mb_per_second = bytes_per_second / (1024 * 1024)

    # Calculate time to fill max file size
    time_to_fill = max_file_size_mb / mb_per_second

    # Determine optimal window
    if time_to_fill < min_time_window_sec:
        # High throughput - use size-based
        return {
            "strategy": "size_based",
            "size_limit_mb": max_file_size_mb,
            "time_window_sec": min_time_window_sec,
            "expected_files_per_hour": 3600 / min_time_window_sec
        }
    elif time_to_fill > max_time_window_sec:
        # Low throughput - use time-based
        return {
            "strategy": "time_based",
            "size_limit_mb": max_file_size_mb,
            "time_window_sec": max_time_window_sec,
            "expected_files_per_hour": 3600 / max_time_window_sec
        }
    else:
        # Balanced
        return {
            "strategy": "balanced",
            "size_limit_mb": max_file_size_mb,
            "time_window_sec": int(time_to_fill),
            "expected_files_per_hour": 3600 / time_to_fill
        }

# Example
config = calculate_optimal_capture_window(
    events_per_second=1000,
    avg_event_size_bytes=1024
)
print(f"Recommended strategy: {config['strategy']}")
print(f"Time window: {config['time_window_sec']} seconds")

# Organize by business dimensions in addition to time
def generate_custom_path_format(business_dimensions):
    """Generate capture path with business partitioning."""
    base_format = "{Namespace}/{EventHub}"

    # Add business dimensions
    for dimension in business_dimensions:
        base_format += f"/{dimension}={{{dimension}}}"

    # Add time partitions
    base_format += "/year={Year}/month={Month}/day={Day}/hour={Hour}"
    base_format += "/{PartitionId}_{Year}_{Month}_{Day}_{Hour}_{Minute}_{Second}"

    return base_format

# Example: Partition by region and device type
custom_format = generate_custom_path_format(["region", "device_type"])
print(custom_format)
# Output: {Namespace}/{EventHub}/region={region}/device_type={device_type}/year={Year}/month={Month}/day={Day}/hour={Hour}/{PartitionId}_{Year}_{Month}_{Day}_{Hour}_{Minute}_{Second}

from delta.tables import DeltaTable
from pyspark.sql.functions import col, from_json, to_timestamp

# Define checkpoint location
checkpoint_path = "/mnt/checkpoints/captured_events"

# Read captured files as stream
captured_stream = (
    spark.readStream
    .format("avro")
    .load("abfss://eventhub-capture@capturestorage.dfs.core.windows.net/mynamespace/telemetry-events/*/*/*/*/*/*/*.avro")
)

# Parse and transform
schema = "sensor_id STRING, temperature DOUBLE, timestamp STRING"
processed_stream = (
    captured_stream
    .withColumn("parsed_body", from_json(col("Body").cast("string"), schema))
    .withColumn("event_timestamp", to_timestamp(col("EnqueuedTimeUtc")))
    .select(
        "event_timestamp",
        "PartitionId",
        "SequenceNumber",
        "parsed_body.*"
    )
)

# Write to Delta Lake with deduplication
(
    processed_stream.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", checkpoint_path)
    .option("mergeSchema", "true")
    .start("/mnt/delta/telemetry_events")
)