# .github/workflows/spark_pipeline_ci.yml

name: Spark Pipeline CI/CD       # Display name in GitHub UI

# ── TRIGGERS ──────────────────────────────────────
on:
  push:
    branches: [main, develop]      # Run on push to these branches
  pull_request:
    branches: [main]               # Run on PRs targeting main
  schedule:
    - cron: '0 6 * * 1'            # Every Monday at 6am UTC

# ── JOBS ──────────────────────────────────────────
jobs:
  ci:                              # Job ID
    runs-on: ubuntu-latest         # Runner environment
    env:
      JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64

    steps:
      - name: Checkout code
        uses: actions/checkout@v4

      - name: Set up Python 3.11
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Set up Java (Spark needs Java)
        uses: actions/setup-java@v4
        with:
          distribution: 'temurin'
          java-version: '11'

      - name: Install Python dependencies
        run: |
          pip install --upgrade pip
          pip install -r requirements.txt

      - name: Lint with flake8
        run: flake8 src/ --max-line-length=120

      - name: Format check with black
        run: black --check src/

      - name: Run unit tests
        run: pytest tests/ -v --tb=short

      - name: Build wheel package
        run: python setup.py bdist_wheel

      - name: Upload wheel artifact
        uses: actions/upload-artifact@v4
        with:
          name: pipeline-wheel
          path: dist/*.whl

# tests/conftest.py
import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def spark():
    """Create a local SparkSession for the entire test session."""
    spark = (
        SparkSession.builder
        .master("local[2]")       # 2 local threads
        .appName("CITests")
        .config("spark.sql.shuffle.partitions", "2")  # small for tests
        .getOrCreate()
    )
    yield spark
    spark.stop()

# tests/test_transforms.py
from src.transforms import clean_customer_data

def test_null_email_removed(spark):
    # Arrange
    data = [(1, "Alice", "alice@test.com"),
            (2, "Bob",   None)]           # null email — should be dropped
    df = spark.createDataFrame(data, ["id", "name", "email"])

    # Act
    result = clean_customer_data(df)

    # Assert
    assert result.count() == 1
    assert result.first()["name"] == "Alice"

# Install linting tools
pip install flake8 black

# Check for style errors (E: errors, W: warnings, F: pyflakes)
flake8 src/ --max-line-length=120 --ignore=E203,W503

# Check formatting (--check means "would black change this?" — exits non-zero if yes)
black --check src/

# Auto-format (run locally, not in CI — CI just checks)
black src/

# .flake8 config file in project root
cat .flake8
# [flake8]
# max-line-length = 120
# ignore = E203, W503
# exclude = .git, __pycache__, dist

build-and-publish:
  needs: [ci]                     # only runs if CI job passes
  runs-on: ubuntu-latest
  if: github.ref == 'refs/heads/main'  # only on main branch

  steps:
    - uses: actions/checkout@v4

    - name: Build wheel
      run: python setup.py bdist_wheel

    - name: Configure AWS credentials
      uses: aws-actions/configure-aws-credentials@v4
      with:
        aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
        aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
        aws-region: us-east-1

    - name: Upload wheel to S3
      run: |
        VERSION=$(python setup.py --version)
        aws s3 cp dist/*.whl \
          s3://my-artifacts/pyspark-pipelines/v${VERSION}/

deploy-databricks:
  needs: [build-and-publish]
  runs-on: ubuntu-latest

  steps:
    - uses: actions/checkout@v4

    - name: Install Databricks CLI
      run: pip install databricks-cli

    - name: Configure Databricks CLI
      env:
        DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
        DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
      run: |
        # Configure the CLI with workspace URL and PAT token
        databricks configure --token <<EOF
        $DATABRICKS_HOST
        $DATABRICKS_TOKEN
        EOF

    - name: Upload wheel to DBFS
      run: databricks fs cp dist/*.whl dbfs:/libraries/ --overwrite

    - name: Deploy using Asset Bundles
      run: |
        databricks bundle deploy --target prod
        # Reads databricks.yml in the repo root
        # Deploys jobs, pipelines, notebooks defined there

env:
  # ✅ Correct — reference from secrets store
  DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
  AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
  SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}

# Set secrets: GitHub repo → Settings → Secrets → New repository secret
# They are masked in all logs automatically

# .gitlab-ci.yml — PySpark Pipeline CI/CD

# Define all stages (run top to bottom)
stages:
  - lint
  - test
  - build
  - deploy-dev
  - deploy-prod

# Global variables available to all jobs
variables:
  PYTHON_VERSION: "3.11"
  PYSPARK_VERSION: "3.5.0"

# ── STAGE: lint ──────────────────────────────────
lint-code:
  stage: lint
  image: python:3.11-slim
  script:
    - pip install flake8 black
    - flake8 src/ --max-line-length=120
    - black --check src/
  only:
    - merge_requests          # Run on every MR
    - main

# ── STAGE: test ──────────────────────────────────
unit-tests:
  stage: test
  image: python:3.11
  before_script:
    - apt-get update && apt-get install -y default-jdk
    - pip install pyspark==3.5.0 pytest pytest-cov
    - pip install -r requirements.txt
  script:
    - pytest tests/ -v --cov=src --cov-report=xml
  coverage: '/TOTAL.*\s+(\d+%)$/'  # Parse coverage from output
  artifacts:
    reports:
      coverage_report:
        coverage_format: cobertura
        path: coverage.xml

# ── STAGE: build ──────────────────────────────────
build-wheel:
  stage: build
  image: python:3.11
  script:
    - pip install wheel setuptools
    - python setup.py bdist_wheel
  artifacts:
    paths:
      - dist/*.whl           # Store the wheel for downstream jobs
    expire_in: 30 days
  only:
    - main

# ── STAGE: deploy-dev ──────────────────────────────
deploy-to-dev:
  stage: deploy-dev
  image: amazon/aws-cli
  script:
    - aws s3 cp dist/*.whl s3://artifacts-dev/pipelines/
    - databricks bundle deploy --target dev
  environment:
    name: development
  only:
    - main

# ── STAGE: deploy-prod ─────────────────────────────
deploy-to-prod:
  stage: deploy-prod
  image: amazon/aws-cli
  script:
    - aws s3 cp dist/*.whl s3://artifacts-prod/pipelines/
    - databricks bundle deploy --target prod
  environment:
    name: production
  when: manual             # Requires a human to click "Deploy" in GitLab UI
  only:
    - main

# Install GitLab Runner on your server or EC2 instance
curl -L https://packages.gitlab.com/install/repositories/runner/gitlab-runner/script.deb.sh | sudo bash
sudo apt-get install gitlab-runner

# Register it with your GitLab instance
sudo gitlab-runner register \
  --url https://gitlab.com \
  --registration-token YOUR_TOKEN \
  --executor docker \
  --docker-image python:3.11 \
  --description "spark-runner" \
  --tag-list spark,pyspark         # Jobs can target this runner by tag

# In .gitlab-ci.yml, use this runner with tags:
# tags:
#   - spark

# Build stage creates the artifact
build-wheel:
  stage: build
  script:
    - python setup.py bdist_wheel
  artifacts:
    paths:
      - dist/                # Persist this directory
    expire_in: 7 days       # Auto-delete after 7 days

# Deploy stage automatically has access to dist/ from build stage
deploy-to-dev:
  stage: deploy-dev
  dependencies:
    - build-wheel            # Download artifacts from this job
  script:
    - ls dist/               # Can access the .whl file built above
    - aws s3 cp dist/*.whl s3://artifacts-dev/

deploy-emr:
  stage: deploy-prod
  image: amazon/aws-cli
  variables:
    AWS_DEFAULT_REGION: us-east-1
  script:
    - # Upload PySpark script and wheel to S3
    - aws s3 cp src/main_pipeline.py s3://my-emr-scripts/
    - aws s3 cp dist/*.whl s3://my-emr-scripts/
    - # Submit an EMR step
    - |
      aws emr add-steps \
        --cluster-id $CLUSTER_ID \
        --steps Type=Spark,Name="DailyPipeline",\
                Args=[--deploy-mode,cluster,\
                      --py-files,s3://my-emr-scripts/my_pipeline.whl,\
                      s3://my-emr-scripts/main_pipeline.py],\
                ActionOnFailure=CONTINUE
  environment:
    name: production
  when: manual

// Jenkinsfile — PySpark Pipeline Build & Deploy

pipeline {
    agent {
        docker {
            image 'python:3.11'          // Run all stages in this Docker image
            args  '-v /var/run/docker.sock:/var/run/docker.sock'
        }
    }

    environment {
        DATABRICKS_HOST  = credentials('databricks-host')   // Jenkins credential store
        DATABRICKS_TOKEN = credentials('databricks-token')
        AWS_CREDENTIALS  = credentials('aws-de-credentials') // UsernamePassword binding
    }

    stages {
        stage('Checkout') {
            steps {
                checkout scm                // Pull code from Git
            }
        }

        stage('Install Dependencies') {
            steps {
                sh '''
                    pip install --upgrade pip
                    pip install -r requirements.txt
                    apt-get install -y default-jdk -q
                '''
            }
        }

        stage('Lint') {
            steps {
                sh 'flake8 src/ --max-line-length=120'
                sh 'black --check src/'
            }
        }

        stage('Unit Tests') {
            steps {
                sh 'pytest tests/ -v --junitxml=test-results.xml'
            }
            post {
                always {
                    junit 'test-results.xml'    // Publish test results in Jenkins UI
                }
            }
        }

        stage('Build Wheel') {
            steps {
                sh 'python setup.py bdist_wheel'
                archiveArtifacts artifacts: 'dist/*.whl'  // Store in Jenkins
            }
        }

        stage('Deploy to Dev') {
            when {
                branch 'main'              // Only on main branch
            }
            steps {
                sh '''
                    databricks fs cp dist/*.whl dbfs:/libraries/ --overwrite
                    databricks jobs run-now --job-id $DEV_JOB_ID
                '''
            }
        }

        stage('Deploy to Prod') {
            when {
                branch 'main'
            }
            input {
                message "Deploy to PRODUCTION?"  // Human approval gate
                ok "Yes, deploy"
            }
            steps {
                sh 'databricks bundle deploy --target prod'
            }
        }
    }

    post {
        failure {
            emailext subject: "Pipeline FAILED: ${env.JOB_NAME} #${env.BUILD_NUMBER}",
                     body:    "Check Jenkins: ${env.BUILD_URL}",
                     to:      'data-team@company.com'
        }
        success {
            echo 'Pipeline completed successfully!'
        }
    }
}

// Use withAWS() from the Pipeline: AWS Steps plugin
stage('Upload to S3') {
    steps {
        withAWS(credentials: 'aws-de-credentials', region: 'us-east-1') {
            s3Upload(
                file:   'dist/my_pipeline-1.0.0-py3-none-any.whl',
                bucket: 'my-artifacts',
                path:   'pyspark-pipelines/'
            )
        }
    }
}

// Trigger an EMR step
stage('Run EMR Job') {
    steps {
        withAWS(credentials: 'aws-de-credentials', region: 'us-east-1') {
            sh """
                aws emr add-steps \
                  --cluster-id j-XXXXXXXXXXX \
                  --steps Type=Spark,Name=DailyETL,\
                          ActionOnFailure=CONTINUE,\
                          Args=[--deploy-mode,cluster,\
                                s3://my-scripts/main.py]
            """
        }
    }
}

my_pipeline/                  # Project root
├── setup.py                  # or pyproject.toml
├── requirements.txt
├── src/
│   └── my_pipeline/
│       ├── __init__.py
│       ├── transforms.py     # Your PySpark code
│       ├── utils.py
│       └── config.py
└── tests/
    ├── conftest.py
    └── test_transforms.py

# setup.py
from setuptools import setup, find_packages

setup(
    name="my_pipeline",
    version="1.4.2",
    description="Customer data pipeline",
    author="Data Engineering Team",
    python_requires=">=3.9",
    packages=find_packages(where="src"),
    package_dir={"": "src"},
    install_requires=[
        "pyspark==3.5.0",
        "delta-spark==3.1.0",
        "boto3>=1.34.0",
    ],
    extras_require={
        "dev": ["pytest", "black", "flake8"]
    }
)

# pyproject.toml — modern Python packaging
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.backends.legacy:build"

[project]
name = "my_pipeline"
version = "1.4.2"
description = "Customer data pipeline"
requires-python = ">=3.9"
dependencies = [
    "pyspark==3.5.0",
    "delta-spark==3.1.0",
    "boto3>=1.34.0",
]

[project.optional-dependencies]
dev = ["pytest", "black", "flake8"]

[tool.setuptools.packages.find]
where = ["src"]

# Build using setuptools (setup.py way)
python setup.py bdist_wheel
# Output: dist/my_pipeline-1.4.2-py3-none-any.whl

# Build using build module (pyproject.toml way — modern)
pip install build
python -m build
# Output: dist/my_pipeline-1.4.2-py3-none-any.whl

# Install the wheel to verify it works
pip install dist/my_pipeline-1.4.2-py3-none-any.whl
python -c "from my_pipeline.transforms import clean_customer_data; print('OK')"

# Upload wheel to S3 artifact store
aws s3 cp dist/my_pipeline-1.4.2-py3-none-any.whl \
    s3://my-artifacts/pyspark-pipelines/

# In spark-submit — distribute wheel to all executors
spark-submit \
    --py-files s3://my-artifacts/pyspark-pipelines/my_pipeline-1.4.2-py3-none-any.whl \
    main.py

# In Databricks — install as a cluster library
databricks libraries install \
    --cluster-id 0101-XXXXX \
    --whl s3://my-artifacts/pyspark-pipelines/my_pipeline-1.4.2-py3-none-any.whl

deploy-dev:
  needs: [unit-tests]
  runs-on: ubuntu-latest
  if: github.ref == 'refs/heads/develop'
  environment: development          # GitHub environment
  env:
    DATABRICKS_HOST: ${{ secrets.DEV_DATABRICKS_HOST }}
    DATABRICKS_TOKEN: ${{ secrets.DEV_DATABRICKS_TOKEN }}
  steps:
    - uses: actions/checkout@v4
    - name: Deploy to Dev workspace
      run: databricks bundle deploy --target dev

# tests/integration/test_pipeline_qa.py
# Run in QA environment with real data connections

import pytest
from pyspark.sql import SparkSession
from my_pipeline.transforms import run_customer_pipeline

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder.appName("QA-Integration").getOrCreate()

def test_pipeline_row_count(spark):
    """Validate output row count within expected range."""
    result = run_customer_pipeline(
        spark,
        input_path="s3://qa-data/customers/2024-01-01/",
        output_path="s3://qa-output/customers/"
    )
    row_count = result.count()
    assert 900_000 < row_count < 1_100_000, \
        f"Unexpected row count: {row_count}"

def test_no_null_customer_ids(spark):
    """Validate no nulls in primary key."""
    result = run_customer_pipeline(
        spark,
        input_path="s3://qa-data/customers/2024-01-01/"
    )
    null_count = result.filter(result.customer_id.isNull()).count()
    assert null_count == 0, "Found null customer_ids!"

# To rollback: redeploy the previous wheel version

# 1. Find the previous version in S3
aws s3 ls s3://my-artifacts/pyspark-pipelines/
#   my_pipeline-1.4.1-py3-none-any.whl   ← previous good version
#   my_pipeline-1.4.2-py3-none-any.whl   ← broken version (rollback from this)

# 2. Install the previous version on the cluster
databricks libraries install \
    --cluster-id 0101-PROD-XXXXX \
    --whl s3://my-artifacts/pyspark-pipelines/my_pipeline-1.4.1-py3-none-any.whl

# 3. Or redeploy the git tag of the previous release
git checkout v1.4.1
databricks bundle deploy --target prod

# main.tf — Provision a Databricks job cluster

terraform {
  required_providers {
    databricks = {
      source  = "databricks/databricks"
      version = "~> 1.40"
    }
  }
  # Remote state in S3 (never use local state in team settings)
  backend "s3" {
    bucket         = "my-terraform-state"
    key            = "pyspark-pipelines/terraform.tfstate"
    region         = "us-east-1"
    dynamodb_table = "terraform-lock"    # Prevents concurrent applies
  }
}

provider "databricks" {
  host  = var.databricks_host
  token = var.databricks_token
}

# Create a Databricks Job with a job cluster
resource "databricks_job" "customer_pipeline" {
  name = "customer-data-pipeline"

  new_cluster {
    num_workers   = 4
    spark_version = "14.3.x-scala2.12"
    node_type_id  = "i3.xlarge"

    spark_conf = {
      "spark.sql.shuffle.partitions" = "200"
      "spark.databricks.delta.optimizeWrite.enabled" = "true"
    }

    aws_attributes {
      instance_profile_arn = var.instance_profile_arn
      availability          = "SPOT_WITH_FALLBACK"
    }
  }

  spark_python_task {
    python_file = "s3://my-scripts/main_pipeline.py"
    parameters  = ["--env", "prod"]
  }

  library {
    whl = "s3://my-artifacts/pyspark-pipelines/my_pipeline-1.4.2-py3-none-any.whl"
  }

  schedule {
    quartz_cron_expression = "0 0 6 * * ?"   # Every day at 6am UTC
    timezone_id            = "UTC"
  }

  email_notifications {
    on_failure = ["data-team@company.com"]
  }
}

# EMR cluster for PySpark jobs
resource "aws_emr_cluster" "spark_cluster" {
  name          = "data-pipeline-cluster"
  release_label = "emr-6.15.0"
  applications  = ["Spark", "Hadoop"]

  ec2_attributes {
    subnet_id                         = var.private_subnet_id
    emr_managed_master_security_group = aws_security_group.emr_master.id
    emr_managed_slave_security_group  = aws_security_group.emr_slave.id
    instance_profile                  = aws_iam_instance_profile.emr.arn
  }

  master_instance_group {
    instance_type = "m5.xlarge"
  }

  core_instance_group {
    instance_type  = "m5.2xlarge"
    instance_count = 4

    ebs_config {
      size                 = 100
      type                 = "gp2"
      volumes_per_instance = 1
    }
  }

  configurations_json = jsonencode([{
    Classification = "spark-defaults"
    Properties = {
      "spark.sql.shuffle.partitions" = "200"
      "spark.executor.memory"        = "8g"
    }
  }])

  service_role = aws_iam_role.emr_service.arn
  auto_termination_policy { idle_timeout = 3600 }

  tags = { Environment = var.environment }
}

# S3 bucket for pipeline artifacts
resource "aws_s3_bucket" "artifacts" {
  bucket = "my-company-pipeline-artifacts-${var.environment}"
}
resource "aws_s3_bucket_versioning" "artifacts" {
  bucket = aws_s3_bucket.artifacts.id
  versioning_configuration { status = "Enabled" }
}

# Initialize — download providers, configure backend
terraform init

# Plan — show what will be created/changed/destroyed
terraform plan -out=tfplan

# Apply — create/update infrastructure
terraform apply tfplan

# Destroy — tear down all resources (careful!)
terraform destroy

# Workspaces for dev/staging/prod
terraform workspace new dev
terraform workspace new prod
terraform workspace select prod
terraform apply -var="environment=prod"

# Create an annotated Git tag for a release
git tag -a v1.5.0 -m "feat: add country_code to customer output"

# Push the tag to GitHub (triggers release workflow)
git push origin v1.5.0

# List all release tags
git tag --list 'v*'

# Get the current version from git tag in CI
VERSION=$(git describe --tags --abbrev=0 | sed 's/v//')
echo "Building version: $VERSION"   # e.g. 1.5.0

# Changelog

## [1.5.0] - 2024-03-15
### Added
- New `country_code` column derived from phone number normalization
- Support for reading from MSK (Managed Kafka) as source

### Changed
- Improved join performance using broadcast hint for country dim table

## [1.4.3] - 2024-03-10
### Fixed
- Fixed NullPointerException when customer email field is missing
- Fixed incorrect SLA timestamp in audit table

## [1.4.2] - 2024-03-01
### Changed
- Upgraded PySpark from 3.4.0 to 3.5.0
- Reduced shuffle partitions from 400 to 200 (20% faster)

## [1.4.0] - 2024-02-15
### Added
- SCD Type 2 support for customer dimension
- Data Quality checks using Great Expectations

## [2.0.0] - Breaking
### BREAKING CHANGES
- Removed deprecated `cust_email` column (use `email` instead)
- Changed partition scheme from `created_date` to `event_date`

# .github/workflows/release.yml
name: Create Release

on:
  push:
    tags:
      - 'v*'                  # Trigger on any version tag

jobs:
  release:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Build wheel
        run: python setup.py bdist_wheel

      - name: Create GitHub Release
        uses: softprops/action-gh-release@v2
        with:
          files: dist/*.whl
          generate_release_notes: true   # Auto-generate from PR titles
          token: ${{ secrets.GITHUB_TOKEN }}