Reproduction Guide

"""Download 24 months of NYC TLC yellow taxi parquet from CloudFront and stage to S3."""
from __future__ import annotations
import concurrent.futures as cf
import subprocess
import sys
import time
from pathlib import Path
from urllib.request import urlretrieve

# EDIT THESE TWO LINES FOR YOUR ACCOUNT:
BUCKET = "sparkrules-bench-dev-artifacts-YOURACCOUNTID"
REGION = "us-east-1"

S3_PREFIX = "input/yellow/"
CLOUDFRONT = "https://d37ci6vzurychx.cloudfront.net/trip-data/"
LOCAL_DIR = Path("./yellow_cache")
LOCAL_DIR.mkdir(parents=True, exist_ok=True)

MONTHS = [f"{y}-{m:02d}" for y in (2023, 2024) for m in range(1, 13)]

def fetch_and_stage(month):
    fname = f"yellow_tripdata_{month}.parquet"
    local = LOCAL_DIR / fname
    if not local.exists():
        urlretrieve(f"{CLOUDFRONT}{fname}", local)
    size = local.stat().st_size
    subprocess.run(
        ["aws", "s3", "cp", str(local), f"s3://{BUCKET}/{S3_PREFIX}{fname}",
         "--region", REGION, "--only-show-errors"],
        check=True,
    )
    return (month, size)

print(f"staging {len(MONTHS)} months to s3://{BUCKET}/{S3_PREFIX}")
with cf.ThreadPoolExecutor(max_workers=6) as ex:
    for month, size in ex.map(fetch_and_stage, MONTHS):
        print(f"  [{month}] {size/1_000_000:.1f} MB")
print("done")

terraform {
  required_version = ">= 1.6.0"
  required_providers {
    aws = { source = "hashicorp/aws", version = "~> 5.0" }
  }
}

provider "aws" {
  region = var.region
}

locals {
  name_prefix = "${var.project}-${var.environment}"
  tags = {
    Project     = var.project
    Environment = var.environment
    ManagedBy   = "terraform"
  }
}

# --- S3 buckets ---

resource "aws_s3_bucket" "artifacts" {
  bucket        = "${local.name_prefix}-artifacts-${data.aws_caller_identity.current.account_id}"
  force_destroy = true
  tags          = local.tags
}

resource "aws_s3_bucket_versioning" "artifacts" {
  bucket = aws_s3_bucket.artifacts.id
  versioning_configuration { status = "Enabled" }
}

resource "aws_s3_bucket_public_access_block" "artifacts" {
  bucket                  = aws_s3_bucket.artifacts.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "aws_s3_bucket_server_side_encryption_configuration" "artifacts" {
  bucket = aws_s3_bucket.artifacts.id
  rule { apply_server_side_encryption_by_default { sse_algorithm = "AES256" } }
}

resource "aws_s3_bucket" "results" {
  bucket        = "${local.name_prefix}-results-${data.aws_caller_identity.current.account_id}"
  force_destroy = true
  tags          = local.tags
}

resource "aws_s3_bucket_public_access_block" "results" {
  bucket                  = aws_s3_bucket.results.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "aws_s3_bucket_server_side_encryption_configuration" "results" {
  bucket = aws_s3_bucket.results.id
  rule { apply_server_side_encryption_by_default { sse_algorithm = "AES256" } }
}

# --- Upload the Glue scripts + DRL pack ---

resource "aws_s3_object" "glue_job_script" {
  bucket = aws_s3_bucket.artifacts.id
  key    = "scripts/benchmark_taxi_rules.py"
  source = "${path.module}/../glue_job/benchmark_taxi_rules.py"
  etag   = filemd5("${path.module}/../glue_job/benchmark_taxi_rules.py")
}

resource "aws_s3_object" "drl_pack" {
  bucket = aws_s3_bucket.artifacts.id
  key    = "drl/taxi_rules_30.drl"
  source = "${path.module}/../glue_job/taxi_rules_30.drl"
  etag   = filemd5("${path.module}/../glue_job/taxi_rules_30.drl")
}

resource "aws_s3_object" "consolidate_job_script" {
  bucket = aws_s3_bucket.artifacts.id
  key    = "scripts/consolidate_taxi_input.py"
  source = "${path.module}/../glue_job/consolidate_taxi_input.py"
  etag   = filemd5("${path.module}/../glue_job/consolidate_taxi_input.py")
}

# --- IAM role for Glue ---

data "aws_caller_identity" "current" {}

data "aws_iam_policy_document" "glue_assume" {
  statement {
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["glue.amazonaws.com"]
    }
  }
}

resource "aws_iam_role" "glue_job" {
  name               = "${local.name_prefix}-glue-job-role"
  assume_role_policy = data.aws_iam_policy_document.glue_assume.json
  tags               = local.tags
}

resource "aws_iam_role_policy_attachment" "glue_service" {
  role       = aws_iam_role.glue_job.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
}

data "aws_iam_policy_document" "glue_s3" {
  statement {
    sid     = "WriteResults"
    actions = ["s3:PutObject", "s3:GetObject", "s3:DeleteObject", "s3:ListBucket"]
    resources = [
      aws_s3_bucket.results.arn,
      "${aws_s3_bucket.results.arn}/*",
      aws_s3_bucket.artifacts.arn,
      "${aws_s3_bucket.artifacts.arn}/*",
    ]
  }
  statement {
    sid = "Logs"
    actions = ["logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents"]
    resources = ["*"]
  }
}

resource "aws_iam_role_policy" "glue_s3" {
  name   = "${local.name_prefix}-glue-s3"
  role   = aws_iam_role.glue_job.id
  policy = data.aws_iam_policy_document.glue_s3.json
}

# --- Benchmark Glue job ---

resource "aws_glue_job" "benchmark" {
  name              = "${local.name_prefix}-benchmark"
  role_arn          = aws_iam_role.glue_job.arn
  glue_version      = "5.0"
  worker_type       = var.worker_type
  number_of_workers = var.number_of_workers
  timeout           = 30
  max_retries       = 0

  command {
    name            = "glueetl"
    script_location = "s3://${aws_s3_bucket.artifacts.id}/${aws_s3_object.glue_job_script.key}"
    python_version  = "3"
  }

  default_arguments = {
    "--job-language"                     = "python"
    "--enable-metrics"                   = "true"
    "--enable-continuous-cloudwatch-log" = "true"
    "--enable-glue-datacatalog"          = "true"
    "--enable-spark-ui"                  = "true"
    "--spark-event-logs-path"            = "s3://${aws_s3_bucket.artifacts.id}/spark-logs/"
    "--drl_s3_path"                      = "s3://${aws_s3_bucket.artifacts.id}/${aws_s3_object.drl_pack.key}"
    "--input_s3_path"                    = var.input_s3_path
    "--output_s3_path"                   = "s3://${aws_s3_bucket.results.id}/taxi-bench/"
    "--rows_limit"                       = tostring(var.rows_limit)
    "--num_partitions"                   = tostring(var.num_partitions)
    "--TempDir"                          = "s3://${aws_s3_bucket.artifacts.id}/tmp/"
    "--additional-python-modules"        = "sparkrules==${var.sparkrules_version}"
  }

  tags = local.tags
}

# --- Consolidate (one-time schema normalization) Glue job ---

resource "aws_glue_job" "consolidate" {
  name              = "${local.name_prefix}-consolidate"
  role_arn          = aws_iam_role.glue_job.arn
  glue_version      = "5.0"
  worker_type       = var.worker_type
  number_of_workers = var.number_of_workers
  timeout           = 30
  max_retries       = 0

  command {
    name            = "glueetl"
    script_location = "s3://${aws_s3_bucket.artifacts.id}/${aws_s3_object.consolidate_job_script.key}"
    python_version  = "3"
  }

  default_arguments = {
    "--job-language"                     = "python"
    "--enable-metrics"                   = "true"
    "--enable-continuous-cloudwatch-log" = "true"
    "--enable-glue-datacatalog"          = "true"
    "--input_s3_path"                    = "s3://${aws_s3_bucket.artifacts.id}/input/yellow/"
    "--output_s3_path"                   = "s3://${aws_s3_bucket.artifacts.id}/input/yellow_consolidated/"
    "--TempDir"                          = "s3://${aws_s3_bucket.artifacts.id}/tmp/"
  }

  tags = local.tags
}

variable "region"             { type = string  default = "us-east-1" }
variable "project"            { type = string  default = "sparkrules-bench" }
variable "environment"        { type = string  default = "dev" }
variable "worker_type" {
  type = string
  default = "G.1X"
  validation {
    condition     = contains(["G.1X", "G.2X", "G.4X", "G.8X"], var.worker_type)
    error_message = "worker_type must be G.1X, G.2X, G.4X, or G.8X."
  }
}
variable "number_of_workers" {
  type = number
  default = 10
  validation {
    condition     = var.number_of_workers >= 2 && var.number_of_workers <= 200
    error_message = "number_of_workers must be between 2 and 200."
  }
}
variable "input_s3_path"      { type = string  default = "" }
variable "rows_limit"         { type = number  default = 0 }
variable "sparkrules_version" { type = string  default = "1.2.0" }
variable "num_partitions"     { type = number  default = 80 }

output "glue_job_name"        { value = aws_glue_job.benchmark.name }
output "consolidate_job_name" { value = aws_glue_job.consolidate.name }
output "artifacts_bucket"     { value = aws_s3_bucket.artifacts.id }
output "results_bucket"       { value = aws_s3_bucket.results.id }
output "results_path"         { value = "s3://${aws_s3_bucket.results.id}/taxi-bench/" }
output "spark_logs_path"      { value = "s3://${aws_s3_bucket.artifacts.id}/spark-logs/" }
output "start_command" {
  value = "aws glue start-job-run --job-name ${aws_glue_job.benchmark.name} --region ${var.region}"
}

region             = "us-east-1"
project            = "sparkrules-bench"
environment        = "dev"
worker_type        = "G.1X"
number_of_workers  = 10
num_partitions     = 80
sparkrules_version = "1.2.0"
rows_limit         = 0

# Update after first `terraform apply` - replace YOURACCOUNTID with your AWS account id
input_s3_path = "s3://sparkrules-bench-dev-artifacts-YOURACCOUNTID/input/yellow_consolidated/"

"""sparkrules 1.2.0 benchmark on NYC TLC Yellow Taxi data."""
from __future__ import annotations
import json
import sys
import time
from datetime import datetime, timezone

from awsglue.context import GlueContext
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from pyspark.sql import functions as F


def main() -> int:
    args = getResolvedOptions(
        sys.argv,
        ["JOB_NAME", "drl_s3_path", "input_s3_path", "output_s3_path", "rows_limit"],
    )
    try:
        num_partitions = int(getResolvedOptions(sys.argv, ["num_partitions"])["num_partitions"])
    except Exception:
        num_partitions = 80

    sc = SparkContext()
    glue = GlueContext(sc)
    spark = glue.spark_session
    spark.conf.set("spark.sql.adaptive.enabled", "true")
    spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

    drl_text = _read_s3_text(spark, args["drl_s3_path"])

    from sparkrules.compiler.rulepack import RulePack
    from sparkrules.spark.executor import SparkRuleExecutor

    pack = RulePack.from_drl(drl_text)
    summary = pack.summary()

    # Probe the optional Rust backend on the driver
    native_info = {"available": False, "version": None, "error": None}
    try:
        from sparkrules.native.bridge import load_native
        mod = load_native()
        if mod is not None:
            native_info = {"available": True, "version": mod.native_version(), "error": None}
    except Exception as e:
        native_info["error"] = f"{type(e).__name__}: {e}"

    # --- Read NYC TLC input ---
    t_read_start = time.perf_counter()
    df = spark.read.parquet(args["input_s3_path"])
    rows_limit = int(args.get("rows_limit", "0") or "0")
    if rows_limit > 0:
        df = df.limit(rows_limit)

    df = df.select(
        F.struct(
            F.col("VendorID").cast("bigint").alias("vendor_id"),
            F.col("passenger_count").cast("bigint").alias("passenger_count"),
            F.col("trip_distance").cast("double").alias("trip_distance"),
            F.col("fare_amount").cast("double").alias("fare_amount"),
            F.col("tip_amount").cast("double").alias("tip_amount"),
            F.col("total_amount").cast("double").alias("total_amount"),
            F.col("payment_type").cast("bigint").alias("payment_type"),
            F.col("PULocationID").cast("bigint").alias("pu_location"),
            F.col("DOLocationID").cast("bigint").alias("do_location"),
            F.hour(F.col("tpep_pickup_datetime")).alias("pickup_hour"),
            F.dayofweek(F.col("tpep_pickup_datetime")).alias("pickup_dow"),
        ).alias("t")
    ).repartition(num_partitions)
    t_read_elapsed = time.perf_counter() - t_read_start

    # --- Apply rules (Strategy A: Catalyst pushdown) ---
    executor = SparkRuleExecutor.from_drl(drl_text)

    t_scoring_start = time.perf_counter()
    if hasattr(executor, "apply_with_counts"):
        result, counts = executor.apply_with_counts(df)
        t_scoring_elapsed = time.perf_counter() - t_scoring_start
        total = int(counts.pop("_total_rows"))
        fired_counts = {k: int(v) for k, v in counts.items()}
        api_used = "apply_with_counts"
    else:
        result = executor.apply(df)
        rule_cols = [c for c in result.columns if c.startswith("r_")]
        agg_exprs = [F.sum(F.when(F.col(c), 1).otherwise(0)).alias(c) for c in rule_cols]
        agg_exprs.append(F.count(F.lit(1)).alias("total_rows"))
        fired_row = result.agg(*agg_exprs).collect()[0]
        t_scoring_elapsed = time.perf_counter() - t_scoring_start
        total = int(fired_row["total_rows"])
        fired_counts = {c: int(fired_row[c]) for c in rule_cols}
        api_used = "apply+manual_agg"

    rows_per_sec = total / t_scoring_elapsed if t_scoring_elapsed > 0 else 0

    try:
        catalyst_plan = result.limit(1)._jdf.queryExecution().optimizedPlan().toString()[:4000]
    except Exception as e:
        catalyst_plan = f"(plan capture failed: {e})"

    # --- Write typed results ---
    t_write_start = time.perf_counter()
    write_partitions = max(8, num_partitions // 2)
    result.coalesce(write_partitions).write.mode("overwrite").parquet(
        args["output_s3_path"].rstrip("/") + "/results/"
    )
    t_write_elapsed = time.perf_counter() - t_write_start

    # --- Write benchmark summary JSON ---
    bench = {
        "job_name": args["JOB_NAME"],
        "timestamp_utc": datetime.now(timezone.utc).isoformat(),
        "input_s3_path": args["input_s3_path"],
        "rows_scored": total,
        "rows_limit_applied": rows_limit,
        "num_partitions": num_partitions,
        "rulepack_summary": summary,
        "fired_counts": fired_counts,
        "timing": {
            "read_elapsed_s": round(t_read_elapsed, 3),
            "scoring_elapsed_s": round(t_scoring_elapsed, 3),
            "write_elapsed_s": round(t_write_elapsed, 3),
            "rows_per_sec": int(rows_per_sec),
            "api_used": api_used,
        },
        "native_backend": native_info,
        "sparkrules_version": _sparkrules_version(),
        "catalyst_plan_head": catalyst_plan,
    }
    bench_json = json.dumps(bench, indent=2, default=str)
    print(bench_json)

    # Write to S3
    summary_dir = args["output_s3_path"].rstrip("/") + "/benchmark_summary_dir"
    URI = spark._jvm.java.net.URI
    Path = spark._jvm.org.apache.hadoop.fs.Path
    FileSystem = spark._jvm.org.apache.hadoop.fs.FileSystem
    fs = FileSystem.get(URI(summary_dir), spark._jsc.hadoopConfiguration())
    if fs.exists(Path(summary_dir)):
        fs.delete(Path(summary_dir), True)

    rdd = spark.sparkContext.parallelize([bench_json], 1)
    rdd.saveAsTextFile(summary_dir)
    return 0


def _read_s3_text(spark, s3_path: str) -> str:
    return "\n".join(spark.read.text(s3_path).rdd.map(lambda r: r[0]).collect())


def _sparkrules_version() -> str:
    try:
        import sparkrules
        return sparkrules.__version__
    except Exception:
        return "unknown"


if __name__ == "__main__":
    main()

"""Normalize 24 schema-drifted TLC yellow parquets into one consolidated folder."""
from __future__ import annotations
import sys
import time

from awsglue.context import GlueContext
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from pyspark.sql import functions as F


def main():
    args = getResolvedOptions(sys.argv, ["JOB_NAME", "input_s3_path", "output_s3_path"])

    sc = SparkContext()
    glue = GlueContext(sc)
    spark = glue.spark_session
    spark.conf.set("spark.sql.adaptive.enabled", "true")
    spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

    input_path = args["input_s3_path"]
    output_path = args["output_s3_path"]

    URI = spark._jvm.java.net.URI
    Path = spark._jvm.org.apache.hadoop.fs.Path
    FileSystem = spark._jvm.org.apache.hadoop.fs.FileSystem
    fs = FileSystem.get(URI(input_path), spark._jsc.hadoopConfiguration())
    statuses = fs.listStatus(Path(input_path))
    files = sorted(
        s.getPath().toString()
        for s in statuses
        if s.getPath().toString().endswith(".parquet")
    )

    normalize = [
        F.col("VendorID").cast("bigint").alias("VendorID"),
        F.col("tpep_pickup_datetime").cast("timestamp").alias("tpep_pickup_datetime"),
        F.col("tpep_dropoff_datetime").cast("timestamp").alias("tpep_dropoff_datetime"),
        F.col("passenger_count").cast("double").alias("passenger_count"),
        F.col("trip_distance").cast("double").alias("trip_distance"),
        F.col("fare_amount").cast("double").alias("fare_amount"),
        F.col("tip_amount").cast("double").alias("tip_amount"),
        F.col("total_amount").cast("double").alias("total_amount"),
        F.col("payment_type").cast("bigint").alias("payment_type"),
        F.col("PULocationID").cast("bigint").alias("PULocationID"),
        F.col("DOLocationID").cast("bigint").alias("DOLocationID"),
    ]

    t0 = time.perf_counter()
    parts = [spark.read.parquet(f).select(*normalize) for f in files]
    df = parts[0]
    for p in parts[1:]:
        df = df.unionByName(p)

    fs_out = FileSystem.get(URI(output_path), spark._jsc.hadoopConfiguration())
    if fs_out.exists(Path(output_path)):
        fs_out.delete(Path(output_path), True)

    df.repartition(64).write.mode("overwrite").parquet(output_path)
    print(f"[consolidate] wrote in {time.perf_counter() - t0:.2f}s to {output_path}")


if __name__ == "__main__":
    main()

// NYC TLC Yellow Taxi: 30-rule decision pack for the sparkrules cluster benchmark.
// Fact binding: $t : T holds struct fields vendor_id, passenger_count, trip_distance,
// fare_amount, tip_amount, total_amount, payment_type, pu_location, do_location,
// pickup_hour, pickup_dow.
//
// All rules are SQL_PUSHDOWN-classifiable (V2 Strategy A) so the benchmark
// measures pure Catalyst predicate throughput.

rule "r01_fare_zero_or_negative"
  salience 100
  when $t : T( $t.fare_amount <= 0 )
  then result.flag = "DATA_QUALITY"; result.reason = "FARE_NONPOSITIVE"; end

rule "r02_fare_extreme_high"
  salience 90
  when $t : T( $t.fare_amount > 500 )
  then result.flag = "REVIEW"; result.reason = "FARE_EXTREME"; end

rule "r03_distance_zero_nonzero_fare"
  salience 95
  when $t : T( $t.trip_distance == 0 and $t.fare_amount > 10 )
  then result.flag = "ANOMALY"; result.reason = "ZERO_DIST_POSITIVE_FARE"; end

rule "r04_distance_extreme"
  salience 85
  when $t : T( $t.trip_distance > 100 )
  then result.flag = "REVIEW"; result.reason = "DISTANCE_OVER_100MI"; end

rule "r05_tip_exceeds_fare"
  salience 90
  when $t : T( $t.tip_amount > $t.fare_amount and $t.fare_amount > 0 )
  then result.flag = "REVIEW"; result.reason = "TIP_EXCEEDS_FARE"; end

rule "r06_tip_negative"
  salience 100
  when $t : T( $t.tip_amount < 0 )
  then result.flag = "DATA_QUALITY"; result.reason = "TIP_NEGATIVE"; end

rule "r07_high_tip_percentage"
  salience 50
  when $t : T( $t.tip_amount > 20 and $t.fare_amount < 30 )
  then result.flag = "GENEROUS_TIP"; result.reason = "TIP_RATIO_HIGH"; end

rule "r08_cash_tip_unusual"
  salience 80
  when $t : T( $t.payment_type == 2 and $t.tip_amount > 0 )
  then result.flag = "ANOMALY"; result.reason = "CASH_TIP_RECORDED"; end

rule "r09_short_trip_high_fare"
  salience 85
  when $t : T( $t.trip_distance < 0.5 and $t.fare_amount > 40 )
  then result.flag = "ANOMALY"; result.reason = "SHORT_TRIP_HIGH_FARE"; end

rule "r10_long_trip_low_fare"
  salience 80
  when $t : T( $t.trip_distance > 20 and $t.fare_amount < 30 )
  then result.flag = "REVIEW"; result.reason = "LONG_TRIP_LOW_FARE"; end

rule "r11_passenger_zero"
  salience 95
  when $t : T( $t.passenger_count == 0 )
  then result.flag = "DATA_QUALITY"; result.reason = "ZERO_PASSENGERS"; end

rule "r12_passenger_over_six"
  salience 80
  when $t : T( $t.passenger_count > 6 )
  then result.flag = "ANOMALY"; result.reason = "PASSENGER_COUNT_HIGH"; end

rule "r13_airport_jfk_pickup"
  salience 30
  when $t : T( $t.pu_location == 132 )
  then result.flag = "AIRPORT_TRIP"; result.reason = "PICKUP_JFK"; end

rule "r14_airport_lga_pickup"
  salience 30
  when $t : T( $t.pu_location == 138 )
  then result.flag = "AIRPORT_TRIP"; result.reason = "PICKUP_LGA"; end

rule "r15_airport_ewr_pickup"
  salience 30
  when $t : T( $t.pu_location == 1 )
  then result.flag = "AIRPORT_TRIP"; result.reason = "PICKUP_EWR"; end

rule "r16_rush_hour_am"
  salience 20
  when $t : T( $t.pickup_hour >= 7 and $t.pickup_hour <= 9 )
  then result.flag = "RUSH_HOUR"; result.reason = "AM_RUSH"; end

rule "r17_rush_hour_pm"
  salience 20
  when $t : T( $t.pickup_hour >= 17 and $t.pickup_hour <= 19 )
  then result.flag = "RUSH_HOUR"; result.reason = "PM_RUSH"; end

rule "r18_late_night"
  salience 15
  when $t : T( $t.pickup_hour >= 0 and $t.pickup_hour <= 4 )
  then result.flag = "LATE_NIGHT"; result.reason = "LATE_NIGHT_TRIP"; end

rule "r19_weekend_trip"
  salience 10
  when $t : T( $t.pickup_dow == 1 or $t.pickup_dow == 7 )
  then result.flag = "WEEKEND"; result.reason = "WEEKEND_TRIP"; end

rule "r20_vendor_cmt"
  salience 5
  when $t : T( $t.vendor_id == 1 )
  then result.flag = "VENDOR"; result.reason = "CMT"; end

rule "r21_vendor_vts"
  salience 5
  when $t : T( $t.vendor_id == 2 )
  then result.flag = "VENDOR"; result.reason = "VTS"; end

rule "r22_total_under_fare"
  salience 95
  when $t : T( $t.total_amount < $t.fare_amount )
  then result.flag = "DATA_QUALITY"; result.reason = "TOTAL_LT_FARE"; end

rule "r23_total_extreme"
  salience 85
  when $t : T( $t.total_amount > 1000 )
  then result.flag = "REVIEW"; result.reason = "TOTAL_EXTREME"; end

rule "r24_no_tip_credit"
  salience 40
  when $t : T( $t.payment_type == 1 and $t.tip_amount == 0 and $t.fare_amount > 20 )
  then result.flag = "NOTABLE"; result.reason = "CREDIT_NO_TIP"; end

rule "r25_same_pu_do"
  salience 60
  when $t : T( $t.pu_location == $t.do_location and $t.trip_distance < 0.1 )
  then result.flag = "ANOMALY"; result.reason = "SAME_LOCATION_ZERO_DIST"; end

rule "r26_small_fare_trip"
  salience 5
  when $t : T( $t.fare_amount < 5 )
  then result.flag = "SMALL_FARE"; result.reason = "MINIMUM_FARE_TRIP"; end

rule "r27_medium_trip"
  salience 5
  when $t : T( $t.trip_distance >= 2 and $t.trip_distance <= 5 )
  then result.flag = "MEDIUM_TRIP"; result.reason = "MEDIUM_DISTANCE"; end

rule "r28_long_trip"
  salience 5
  when $t : T( $t.trip_distance > 10 and $t.trip_distance <= 30 )
  then result.flag = "LONG_TRIP"; result.reason = "LONG_DISTANCE"; end

rule "r29_solo_rider"
  salience 5
  when $t : T( $t.passenger_count == 1 )
  then result.flag = "SOLO"; result.reason = "SINGLE_RIDER"; end

rule "r30_group_ride"
  salience 5
  when $t : T( $t.passenger_count >= 4 and $t.passenger_count <= 6 )
  then result.flag = "GROUP"; result.reason = "GROUP_RIDE"; end

# ------ 1. Provision 15 AWS resources ------
cd deploy/aws-glue/terraform/
terraform init
terraform apply -auto-approve

# ------ 2. Capture output names ------
ARTIFACTS_BUCKET=$(terraform output -raw artifacts_bucket)
RESULTS_BUCKET=$(terraform output -raw results_bucket)
BENCH_JOB=$(terraform output -raw glue_job_name)
CONSOLIDATE_JOB=$(terraform output -raw consolidate_job_name)

# ------ 3. Stage 24 months of taxi parquet to S3 (first run only, ~6 min) ------
# Edit stage_taxi_to_s3.py, set BUCKET = $ARTIFACTS_BUCKET, then:
python stage_taxi_to_s3.py

# ------ 4. One-time consolidate (~120s, ~$0.12) ------
aws glue start-job-run --job-name "$CONSOLIDATE_JOB" --region us-east-1

# ------ 5. Run the benchmark (~140s, ~$0.17) ------
aws glue start-job-run --job-name "$BENCH_JOB" --region us-east-1

# Poll for status:
aws glue get-job-run --job-name "$BENCH_JOB" --run-id <id> --region us-east-1 \
  --query "JobRun.{State:JobRunState,ExecTime:ExecutionTime,DPU:DPUSeconds}" \
  --output json

# ------ 6. Pull the summary JSON ------
aws s3 cp "s3://$RESULTS_BUCKET/taxi-bench/benchmark_summary_dir/part-00000" ./bench.json
cat bench.json | jq '.timing, .native_backend'

# ------ 7. Tear down (IMPORTANT - avoid ongoing cost) ------
terraform destroy -auto-approve

# Confirm nothing left:
aws glue list-jobs --region us-east-1 --query "JobNames[?starts_with(@, 'sparkrules')]"
# Expected: []