databricks-parking-tables

name	databricks-parking-tables
description	Use when forming Spark/SQL queries for Databricks parking evaluation - provides table schemas, PySpark patterns, joins, and filter patterns for PUDO/UNPUDO analysis

Databricks Parking Tables

Overview

Reference guide for Databricks tables used in parking evaluation with focus on PUDO/UNPUDO events analysis. Provides schemas, field usage documentation, common joins, PySpark patterns, and filter strategies.

Core principle: All field descriptions include actual usage patterns - what each field filters, joins on, or computes.

When to Use

• Forming Spark/SQL queries for parking evaluation
• Analyzing PUDO (Pick-Up/Drop-Off) and UNPUDO events
• Finding disengagements around parking maneuvers
• Checking AV mode during parking events
• Joining with model nicknames for analysis by model
• Understanding field meanings and usage

Quick Reference

Table	Primary Use	Key Fields
hive_metastore.parking.pudo_unpudo_events	PUDO/UNPUDO parking events	runID, ride_type, pudo_unixus, unpudo_unixus
prod_data_pipeline.raw__model_catalogue_sync.vehicle_run_models	Run to model mapping	run_id, model_session_id
prod_data_pipeline.raw__model_catalogue_sync.model_training_sessions	Model nicknames	id, nickname
prod_data_pipeline.raw__model_catalogue_sync.model_training_session_metadata	Model tags and metadata	model_session_id, tags (JSON)
prod_data_pipeline.raw__model_catalogue_sync.vehicle_runs	Run metadata and tags	run_id, tags (JSON)
prod_data_pipeline.raw__model_catalogue_sync.run_labels	Run time-ranged annotations	run_id, label, start_timestamp_unixus, end_timestamp_unixus
analytics.disengagements	All disengagement events	run_id, timestamp_us, disengagement_what/why
prod_data_pipeline.wayve_corpus.all_data	Vehicle telemetry and sensor data	run_id, timestamp_unixus, automation_active
users__guy_geva.zak_parking_classification	Parking type predictions	run_id, timestamp_unixus, class_name

Common PySpark Patterns

Join PUDO Events with Model Nicknames

CRITICAL: PUDO uses runID (capital I, D), other tables use run_id (lowercase)

from pyspark.sql import functions as F

pudo_df = spark.table("hive_metastore.parking.pudo_unpudo_events")
vehicle_run_models = spark.table("prod_data_pipeline.raw__model_catalogue_sync.vehicle_run_models")
model_sessions = spark.table("prod_data_pipeline.raw__model_catalogue_sync.model_training_sessions")

run_to_session = vehicle_run_models.select("run_id", "model_session_id").dropDuplicates()
session_to_nickname = model_sessions.select(
    F.col("id").alias("model_session_id"), "nickname"
).dropDuplicates()

run_to_model = run_to_session.join(session_to_nickname, on="model_session_id", how="left")
pudo_with_model = pudo_df.join(run_to_model, pudo_df.runID == run_to_model.run_id, "left").drop(run_to_model.run_id)

Find Last Disengagement Before PUDO

from pyspark.sql import functions as F, Window

pudo_df = pudo_df.withColumn("pudo_timestamp", F.to_timestamp(F.col("pudo_unixus") / 1000000))
diseng_df = spark.table("analytics.disengagements")
diseng_df = diseng_df.withColumn("disengagement_timestamp", F.to_timestamp(F.col("timestamp_us") / 1000000))

joined = pudo_df.alias("pudo").join(
    diseng_df.alias("diseng"),
    (F.col("pudo.runID") == F.col("diseng.run_id"))
    & (F.col("diseng.disengagement_timestamp") >= F.col("pudo.pudo_timestamp") - F.expr("INTERVAL 30 SECONDS"))
    & (F.col("diseng.disengagement_timestamp") <= F.col("pudo.pudo_timestamp")),
    "left"
)

window = Window.partitionBy([f"pudo.{c}" for c in pudo_df.columns]).orderBy(F.col("diseng.disengagement_timestamp").desc())
result = joined.withColumn("rank", F.row_number().over(window)).where(
    (F.col("rank") == 1) | F.col("diseng.disengagement_timestamp").isNull()
)

Check AV Mode at PUDO

all_data = spark.table("prod_data_pipeline.wayve_corpus.all_data")
all_data = all_data.withColumn("all_data_timestamp", F.to_timestamp(F.col("timestamp_unixus") / 1000000))

joined = pudo_df.alias("pudo").join(
    all_data.alias("av"),
    (F.col("pudo.runID") == F.col("av.run_id"))
    & (F.col("av.all_data_timestamp") >= F.col("pudo.pudo_timestamp") - F.expr("INTERVAL 2 SECONDS"))
    & (F.col("av.all_data_timestamp") <= F.col("pudo.pudo_timestamp") + F.expr("INTERVAL 2 SECONDS")),
    "left"
)

window = Window.partitionBy([f"pudo.{c}" for c in pudo_df.columns]).orderBy(
    F.abs(F.unix_timestamp(F.col("av.all_data_timestamp")) - F.unix_timestamp(F.col("pudo.pudo_timestamp")))
)

result = joined.withColumn("rank", F.row_number().over(window)).where(
    (F.col("rank") == 1) | F.col("av.all_data_timestamp").isNull()
).withColumn(
    "av_mode_at_pudo",
    F.when(F.col("av.ground_truth__state__vehicle__automation_active").cast("boolean") == F.lit(True), 1).otherwise(0)
)

Filter PUDO Events by Model Tags

from pyspark.sql import functions as F

# Load tables
pudo_df = spark.table("hive_metastore.parking.pudo_unpudo_events")
vehicle_run_models = spark.table("prod_data_pipeline.raw__model_catalogue_sync.vehicle_run_models")
model_metadata = spark.table("prod_data_pipeline.raw__model_catalogue_sync.model_training_session_metadata")

# Join to get model tags
pudo_with_model = pudo_df.join(
    vehicle_run_models,
    pudo_df.runID == vehicle_run_models.run_id,
    "left"
)

pudo_with_tags = pudo_with_model.join(
    model_metadata,
    pudo_with_model.model_session_id == model_metadata.model_session_id,
    "left"
)

# Filter by specific model tag (tags is JSON array)
parking_models = pudo_with_tags.where(
    F.array_contains(F.col("tags"), "parking-model")
)

# Filter by keyed tag
production_models = pudo_with_tags.where(
    F.array_contains(F.col("tags"), "deployment-stage:production")
)

Filter Runs by Run Tags

from pyspark.sql import functions as F

# Load tables
pudo_df = spark.table("hive_metastore.parking.pudo_unpudo_events")
vehicle_runs = spark.table("prod_data_pipeline.raw__model_catalogue_sync.vehicle_runs")

# Join to get run tags
pudo_with_run_tags = pudo_df.join(
    vehicle_runs,
    pudo_df.runID == vehicle_runs.run_id,
    "left"
)

# Filter by run tag (tags is JSON object, not array)
# Example: filter for demo runs
demo_runs = pudo_with_run_tags.where(
    F.get_json_object("tags", "$.run_classification") == "demo"
)

# Or check if a specific tag exists
experiment_runs = pudo_with_run_tags.where(
    F.get_json_object("tags", "$.run_classification") == "experiment"
)

Complete Example: Filter by Both Model and Run Tags

from pyspark.sql import functions as F

# Find PUDO events where:
# - Model is tagged as "parking-model"
# - Run is tagged as "experiment"

pudo_df = spark.table("hive_metastore.parking.pudo_unpudo_events")
vehicle_run_models = spark.table("prod_data_pipeline.raw__model_catalogue_sync.vehicle_run_models")
model_metadata = spark.table("prod_data_pipeline.raw__model_catalogue_sync.model_training_session_metadata")
vehicle_runs = spark.table("prod_data_pipeline.raw__model_catalogue_sync.vehicle_runs")
model_sessions = spark.table("prod_data_pipeline.raw__model_catalogue_sync.model_training_sessions")

# Join everything
result = pudo_df.alias("pudo") \
    .join(vehicle_runs.alias("vr"), F.col("pudo.runID") == F.col("vr.run_id"), "left") \
    .join(vehicle_run_models.alias("vrm"), F.col("pudo.runID") == F.col("vrm.run_id"), "left") \
    .join(model_metadata.alias("mm"), F.col("vrm.model_session_id") == F.col("mm.model_session_id"), "left") \
    .join(model_sessions.alias("ms"), F.col("vrm.model_session_id") == F.col("ms.id"), "left") \
    .where(
        F.array_contains(F.col("mm.tags"), "parking-model") &
        (F.get_json_object(F.col("vr.tags"), "$.run_classification") == "experiment")
    ) \
    .select(
        F.col("pudo.*"),
        F.col("ms.nickname").alias("model_nickname"),
        F.col("mm.tags").alias("model_tags"),
        F.col("vr.tags").alias("run_tags")
    )

Key Reminders

• PUDO table: runID (capital I, D)
• Other tables: run_id (lowercase)
• Model sessions: model_training_sessions.id = vehicle_run_models.model_session_id
• Timestamps: Always divide by 1000000 for microseconds to seconds
• Window functions: Use desc() for LAST, asc() for FIRST, abs(diff) for CLOSEST
• Blacklist filtering: Find nearest FIRST, then apply blacklist
• Model tags: Stored in model_training_session_metadata.tags as JSON array - use F.array_contains() to filter
• Run tags: Stored in vehicle_runs.tags as JSON object - use F.get_json_object() to filter
• Tag format: Model tags are arrays ["tag1", "tag2"], run tags are objects {"run_classification": "demo"}

Tag Conventions

Model Tags:

• Must be lowercase and hyphenated: my-tag-name
• Keyed tags: key:value (e.g., test-purpose:experiment, deployment-stage:production)
• Common examples: parking-model, baseline, production, experimental

Run Tags:

• Stored as JSON object with various keys
• Common keys: run_classification (values: experiment, demo, calibration, etc.)
• Set via console app during experiment creation

See @table-schemas.md for complete field documentation.