Generic Ingestion

Unit 1 of Generic 3-Unit Deployment

ODP Ingestion Pipeline - reads mainframe extracts from GCS and loads to BigQuery.

Flow Diagram

                         Generic INGESTION FLOW
                         ─────────────────

  GCS Landing                  Beam Pipeline                    BigQuery ODP
  ───────────                  ─────────────                    ────────────

  1. .csv splits ──┐
  2. .ok file    ──┼──► Pub/Sub ──► Orchestration ──► 1. Read CSV      ┌──────────────────┐
                   │      Event      (Unit 3)         2. Parse HDR/TRL │ odp_generic.      │
                   │                                  3. Validate      │ - customers       │
                   │                                  4. Add Audit     │ - accounts        │
                   │                                  5. Write to BQ ─►│ - decision        │
                   │                                                   │ - applications    │
                   └──────────────────────────────────┘                └──────────────────┘
                               │
                               ▼
                        ┌─────────────┐
                        │ Archive to  │
                        │ GCS Archive │
                        └─────────────┘

Pattern

JOIN + MAP: 4 entities → 4 ODP tables

Components

Schema Architecture

The ingestion unit maintains Python schema definitions that serve as the single source of truth:

• Purpose: Used by the Beam pipeline for runtime CSV parsing, record-level validation, in-flight PII masking, and BigQuery schema generation.
• Feature: Can dynamically generate BigQuery schemas during job execution (CREATE_IF_NEEDED).
• Location: data_ingestion/schema/ — one Python module per entity (customers, accounts, decision, applications) plus registry.py for entity schema registration.

Library-Driven Ease of Use

The Generic ingestion pipeline is a Lean Consumer of the library framework. It achieves complex mainframe ingestion with minimal custom code by leveraging:

1. Metadata-Driven Schema: data_ingestion/schema/customers.py simply defines an EntitySchema. The library's SchemaValidator handles all type checking and PII masking automatically.
2. Standardized Parsing: Uses the HDRTRLParser from gcp-pipeline-beam to validate mainframe headers/trailers without regex boilerplate.
3. Audit Integrity: Automatically injects _run_id and _processed_at using the AddAuditColumnsDoFn library transform.

How to Replicate this JOIN Ingestion (3-to-3)

To create a new ingestion unit for a multi-entity system, follow the Creating New Deployment Guide.

Key steps for this JOIN pattern:

1. Define Schema: Create a new schema file using gcp_pipeline_core.schema.EntitySchema.
2. Configure Pipeline: Inherit from gcp_pipeline_beam.pipelines.base.BasePipeline.
3. Plug in Transforms: Use the fluent BeamPipelineBuilder to chain read_csv -> validate -> write_to_bigquery.
4. CI/CD Config: Update your project and org identifiers in your CI/CD configuration.

Infrastructure & Configurations

Google Cloud Resources

This deployment requires the following GCP infrastructure, provisioned via Terraform:

• Storage: GCS buckets for landing, archive, and error files.
• Messaging: Pub/Sub Topic generic-file-notifications and Subscription generic-file-notifications-sub.
• Processing: Cloud Dataflow (Apache Beam) for running the ingestion pipeline.
• Data Warehouse: BigQuery dataset odp_generic for raw data storage.

For detailed infrastructure definitions, see infrastructure/terraform/systems/generic/ingestion/.

Pipeline Configuration (GenericPipelineOptions)

The ingestion pipeline accepts several command-line arguments to control its behavior:

Technology Stack & Documentation

• Google Cloud Dataflow - Managed Apache Beam service
• Apache Beam Python SDK - Programming model for data processing
• Google Cloud Storage - Input file landing zone
• Google BigQuery - Data warehouse target
• Google Cloud Pub/Sub - Event-driven triggers
• Apache Beam BigQuery I/O - Connector for BQ

Dependencies

NO Apache Airflow dependency - orchestration is separate unit.

Execution & Testing

1. Local Development Setup

Initialize the virtual environment:

./scripts/setup_deployment_venv.sh original-data-to-bigqueryload
source deployments/original-data-to-bigqueryload/venv/bin/activate

2. Unit Testing

Run the ingestion unit tests:

python -m pytest tests/unit/ -v

3. Local Execution (DirectRunner)

Run the Beam pipeline locally to validate parsing and schema:

python -m data_ingestion.pipeline.runner \
    --input_file=tests/data/sample_customers.csv \
    --output_table=my-project:odp_generic.customers \
    --runner=DirectRunner \
    --temp_location=/tmp/beam-temp

4. Cloud Execution

The ingestion unit is typically triggered by Airflow. To trigger it manually for testing on GCP:

1. Upload a data file to GCS.
2. Upload a corresponding .ok file.
3. The generic_pubsub_trigger_dag will detect the file and launch a Dataflow job using this ingestion code.