[DENG-11196] Port socorro import job from spark

sql/moz-fx-data-shared-prod/telemetry_derived/socorro_crash_v2/README.md

@@ -0,0 +1,60 @@
+# socorro_crash_v2
+
+Daily import of Socorro crash reports from GCS newline JSON into
+`moz-fx-data-shared-prod.telemetry_derived.socorro_crash_v2`, partitioned by `crash_date`.
+
+`query.py` loads one day of JSON from
+`gs://moz-fx-socorro-prod-prod-telemetry/v1/crash_report/<yyyymmdd>/` and writes
+the corresponding `crash_date` partition. This replaces the Spark/Dataproc job
+that used to do this (`https://github.com/mozilla/telemetry-airflow/blob/main/jobs/socorro_import_crash_data.py`).
+
+## How it works
+
+1. Load the day's JSON into a temporary table using a schema derived
+   from `schema.yaml` at runtime
+2. Run `transform.sql` to reshape the temp table into the destination table's
+   layout and write the `crash_date` partition with `WRITE_TRUNCATE`
+
+## Usage
+
+```sh
+python3 query.py --date 2026-01-01
+```
+
+Options: `--date` (required), `--project`, `--source-bucket`, `--source-prefix`,
+`--destination-dataset`, `--destination-table`, `--dry-run`. `--dry-run` prints
+the planned load (source URI, destination partition) without touching BigQuery.
+
+## Migration notes
+
+The original Spark job read GCS JSON, coerced it into a Spark `StructType`
+derived from the JSON (`telemetry_socorro_crash.json`), and wrote
+partitioned parquet to `gs://moz-fx-data-prod-socorro-data/socorro_crash/v2/`. A
+separate `parquet2bigquery` GKE pod then loaded that parquet into this table.
+
+Moving the job here removes the Spark cluster, the intermediate parquet, and the
+separate load job. The load is a single BigQuery load plus one transform query.
+
+### Schema
+
+`schema.yaml` is a copy of the current production table schema, so the
+migrated job targets the existing table shape without redefining it.  The schemas is based on
+Schema is based on https://github.com/mozilla-services/socorro/blob/main/socorro/schemas/telemetry_socorro_crash.json.
+
+Differences from the upstream schema:
+  - `crash_date` (DATE) is the partition column. It is not present in
+    the crash report JSON. The old pipeline derived it from the GCS folder name
+    (`crash_date=<yyyymmdd>`); here it comes from `--date`.
+  - Arrays are stored as `RECORD<list: ARRAY<RECORD<element>>>` rather than
+    plain `REPEATED` fields. This is the encoding Spark's parquet writer
+    produced, and downstream consumers (for example the symbolication jobs that
+    read `json_dump.modules.list`) depend on it. `transform.sql` rewraps the
+    following arrays to match:
+    - `additional_minidumps` (element: STRING)
+    - `addons` (element: STRING)
+    - `json_dump.crashing_thread.frames`
+    - `json_dump.modules`
+    - `json_dump.threads` (and the nested `frames` inside each thread)
+    - `memory_report.reports`
+
+The upstream schema is unlikely to change so it's static instead of dynamically built.

sql/moz-fx-data-shared-prod/telemetry_derived/socorro_crash_v2/metadata.yaml

@@ -1,10 +1,21 @@
-friendly_name: |-
-  Socorro Crash
+friendly_name: Socorro Crash
 description: |-
-  Crash reports imported via socorro.
-  See https://github.com/mozilla/telemetry-airflow/blob/4050697bd2e283864a9013bd48d439eb66f6d581/dags/socorro_import.py#L105
+  Crash reports imported via Socorro.
+
+  Loaded daily from newline JSON crash reports in GCS
+  (gs://moz-fx-socorro-prod-prod-telemetry/v1/crash_report) by query.py. This replaces the
+  Spark/Dataproc job in telemetry-airflow that wrote parquet loaded by parquet2bigquery.
 owners:
-- srose@mozilla.com
-# Schema is imported externally via the socorro_import Airflow DAG; deeply nested
-# crash fields are not individually documented.
+- bewu@mozilla.com
+labels:
+  incremental: true
+  owner1: benwu
+bigquery:
+  time_partitioning:
+    type: day
+    field: crash_date
+    require_partition_filter: true
+    expiration_days: null
+  range_partitioning: null
+  clustering: null
 require_column_descriptions: false

sql/moz-fx-data-shared-prod/telemetry_derived/socorro_crash_v2/query.py

@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+
+"""Load a day of Socorro crash reports from GCS JSON into socorro_crash_v2."""
+
+import os
+import uuid
+from pathlib import Path
+
+import click
+from google.cloud import bigquery, storage
+
+from bigquery_etl.schema import Schema
+
+THIS_DIR = os.path.dirname(__file__)
+DEFAULT_SCHEMA = os.path.join(THIS_DIR, "schema.yaml")
+TRANSFORM_SQL = os.path.join(THIS_DIR, "transform.sql")
+
+
+def load_source_schema():
+    """Build the schema to load the raw JSON with.
+
+    This is the table schema minus crash_date, with the wrapped arrays flattened
+    back to their natural JSON shape. It is derived from schema.yaml at runtime
+    so the load schema and the destination schema never drift apart.
+    """
+    # Use the repo's schema loader so !include-field-description and the other
+    # include tags in schema.yaml are resolved
+    fields = Schema.from_schema_file(Path(DEFAULT_SCHEMA)).schema["fields"]
+    return [
+        bigquery.SchemaField.from_api_repr(_unwrap(field))
+        for field in fields
+        if field["name"] != "crash_date"
+    ]
+
+
+def _unwrap(field):
+    """Flatten a wrapped array back to the plain array the JSON contains.
+
+    A field stored as RECORD<list: ARRAY<RECORD<element>>> becomes the plain
+    REPEATED field the source JSON actually has, recursively.
+    """
+    is_wrapped = field["type"] == "RECORD" and [
+        f["name"] for f in field.get("fields", [])
+    ] == ["list"]
+    if is_wrapped:
+        element = field["fields"][0]["fields"][0]
+        if element["type"] == "RECORD":
+            inner = [_unwrap(f) for f in element.get("fields", [])]
+            return {
+                "name": field["name"],
+                "type": "RECORD",
+                "mode": "REPEATED",
+                "fields": inner,
+            }
+        return {"name": field["name"], "type": element["type"], "mode": "REPEATED"}
+    if field.get("fields"):
+        return {**field, "fields": [_unwrap(f) for f in field["fields"]]}
+    return field
+
+
+@click.command(help=__doc__)
+@click.option(
+    "--date",
+    "date",
+    type=click.DateTime(formats=["%Y-%m-%d"]),
+    required=True,
+    help="Partition date to load, e.g. 2019-08-01.",
+)
+@click.option("--project", default="moz-fx-data-shared-prod")
+@click.option("--source-bucket", default="moz-fx-socorro-prod-prod-telemetry")
+@click.option(
+    "--source-prefix",
+    default="v1/crash_report",
+    help="Source prefix without the date folder.",
+)
+@click.option("--destination-dataset", default="telemetry_derived")
+@click.option("--destination-table", default="socorro_crash_v2")
+@click.option(
+    "--dry-run",
+    is_flag=True,
+    help=(
+        "Smoke test the load: build the load schema, confirm source objects "
+        "exist in GCS, load the temp table, and validate the transform against "
+        "it without writing the destination partition."
+    ),
+)
+def main(
+    date,
+    project,
+    source_bucket,
+    source_prefix,
+    destination_dataset,
+    destination_table,
+    dry_run,
+):
+    """Load one crash_date partition of GCS JSON into BigQuery."""
+    date = date.date()
+
+    date_nodash = date.strftime("%Y%m%d")
+    source_prefix_with_date = f"{source_prefix}/{date_nodash}/"
+    source_uri = f"gs://{source_bucket}/{source_prefix_with_date}*"
+    partition = ".".join(
+        [
+            project,
+            destination_dataset,
+            f"{destination_table}${date_nodash}",
+        ]
+    )
+    with open(TRANSFORM_SQL) as f:
+        transform = f.read()
+
+    client = bigquery.Client(project)
+
+    if dry_run:
+        # Build the load schema locally
+        schema = load_source_schema()
+        print(f"Load schema built: {len(schema)} top-level fields")
+
+        # Confirm the date folder actually holds objects before loading
+        storage_client = storage.Client(project)
+        blobs = storage_client.list_blobs(
+            source_bucket, prefix=source_prefix_with_date, max_results=1
+        )
+        if next(iter(blobs), None) is None:
+            raise click.ClickException(f"No source objects found under {source_uri}")
+        print(f"Source objects present under {source_uri}")
+    else:
+        schema = load_source_schema()
+
+    # Load the raw JSON into a temp table using the natural (unwrapped) schema.
+    tmp_table = f"tmp.socorro_crash_{uuid.uuid4().hex[:8]}"
+    load_result = client.load_table_from_uri(
+        source_uri,
+        tmp_table,
+        location="US",
+        job_config=bigquery.LoadJobConfig(
+            schema=schema,
+            source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
+            write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
+            # Ignore fields present in the JSON but absent from the schema.
+            ignore_unknown_values=True,
+        ),
+    ).result()
+    print(f"Loaded {load_result.output_rows} rows into {tmp_table}")
+
+    try:
+        query = transform.format(source_table=tmp_table, crash_date=date.isoformat())
+
+        if dry_run:
+            # Dry run the transform against the real temp table without writing
+            validate_result = client.query(
+                query,
+                job_config=bigquery.QueryJobConfig(dry_run=True, use_query_cache=False),
+            )
+            print(
+                f"Transform validates; would scan "
+                f"{validate_result.total_bytes_processed} bytes. "
+                f"Skipping write to {partition}."
+            )
+            return
+
+        # Transform into the destination shape and write the partition.
+        query_result = client.query(
+            query,
+            job_config=bigquery.QueryJobConfig(
+                destination=partition,
+                write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
+            ),
+        ).result()
+        print(f"Wrote {query_result.total_rows} rows into {partition}")
+    finally:
+        client.delete_table(tmp_table, not_found_ok=True)
+
+
+if __name__ == "__main__":
+    main()

sql/moz-fx-data-shared-prod/telemetry_derived/socorro_crash_v2/transform.sql

@@ -0,0 +1,54 @@
+-- Transform raw Socorro crash JSON (loaded into a temp table with natural
+-- arrays) into the socorro_crash_v2 shape: inject the crash_date partition
+-- column and rewrap arrays as RECORD<list: ARRAY<RECORD<element>>> to match the
+-- encoding the old Spark parquet writer produced.
+--
+-- {source_table} and {crash_date} are filled in by query.py.
+--
+-- Every scalar column is carried through by name via the * EXCEPT below; only
+-- the seven wrapped arrays and crash_date are rewritten explicitly.
+SELECT
+  DATE("{crash_date}") AS crash_date,
+  * EXCEPT (additional_minidumps, addons, json_dump, memory_report),
+  STRUCT(
+    ARRAY(SELECT AS STRUCT element FROM UNNEST(additional_minidumps) AS element) AS list
+  ) AS additional_minidumps,
+  STRUCT(ARRAY(SELECT AS STRUCT element FROM UNNEST(addons) AS element) AS list) AS addons,
+  (
+    SELECT AS STRUCT
+      json_dump.* EXCEPT (crashing_thread, modules, threads),
+      STRUCT(
+        json_dump.crashing_thread.* EXCEPT (frames),
+        STRUCT(
+          ARRAY(
+            SELECT AS STRUCT
+              element
+            FROM
+              UNNEST(json_dump.crashing_thread.frames) AS element
+          ) AS list
+        ) AS frames
+      ) AS crashing_thread,
+      STRUCT(
+        ARRAY(SELECT AS STRUCT element FROM UNNEST(json_dump.modules) AS element) AS list
+      ) AS modules,
+      STRUCT(
+        ARRAY(
+          SELECT AS STRUCT
+            thread.* EXCEPT (frames),
+            STRUCT(
+              ARRAY(SELECT AS STRUCT element FROM UNNEST(thread.frames) AS element) AS list
+            ) AS frames
+          FROM
+            UNNEST(json_dump.threads) AS thread
+        ) AS list
+      ) AS threads
+  ) AS json_dump,
+  (
+    SELECT AS STRUCT
+      memory_report.* EXCEPT (reports),
+      STRUCT(
+        ARRAY(SELECT AS STRUCT element FROM UNNEST(memory_report.reports) AS element) AS list
+      ) AS reports
+  ) AS memory_report
+FROM
+  `{source_table}`