gcp-dataflow

What this skill does

# Apache Beam Pipelines on Cloud Dataflow

Expert guidance for writing and packaging Apache Beam pipelines to run on Google
Cloud Dataflow.

## Creating a new project

Use this section when creating a new project for a Dataflow pipeline.

-   If the user doesn't say explicitly which language (Java, Python, Go) shall
    be used to write the pipeline, you MUST confirm the language.
-   Determine which version of Beam SDK should be used by searching for the most
    recently released version of Apache Beam, unless the user already uses a
    particular version.
    -   **Action**: Run a web search for the latest Apache Beam SDK release.
-   YOU MUST use same version of Apache Beam consistently throughout the project
    in Dockerfiles, `requirements.txt`, and other similar files where versions
    are specified.

### Java projects using Gradle

Use this section when configuring a Dataflow Java pipeline project using gradle.

-   **Shadow Jars (Fat Jars)**: Do NOT propose to use the Shadow plugin
    (`com.github.johnrengelman.shadow`) unless the user explicitly requests a
    Fat Jar.
-   **Passing command-line parameters**: Use the `application` plugin for
    passing command-line parameters.
-   **SLF4J Logging Dependency Alignment**:
    -   Verify the `slf4j-api` version pulled transitively by Apache Beam.
    -   You MUST configure the application logging backend (`slf4j-simple`,
        `logback-classic`, etc.) to exactly match the major/minor version of the
        resolved `slf4j-api`.

### Structure the pipeline as a Dataflow Flex Template

When creating new Dataflow pipeline projects, configure them as a Flex template.
Flex Templates offer a hermetic and reproducible launch environment, and are
easy to launch with `gcloud` or with orchestrators like Cloud Composer.

Follow the Flex Templates section below.

## Flex Templates

-   **Provide Instructions**: Provide instructions on rebuilding and running
    Flex Templates to the user in walkthrough.
-   **Use Single Docker Image for Python pipelines**: For Python Flex Templates,
    it is better to use a single image for the template launcher image and for
    the worker runtime environment (`--sdk_container_image`). Whenever
    configuring or suggesting a Dataflow Flex Template for a Python pipeline
    that requires extra dependencies (e.g., using `--requirements_file`,
    `--setup_file`, or `--extra_package`), **YOU MUST recommend the Single
    Docker Image Configuration** as detailed in
    [python_flex_template_reference.md](references/python_flex_template_reference.md).
-   **Prefer Cloud Build over Local Docker**:
    -   Do NOT assume local Docker availability on the workspace machine.
    -   **Action**: Suggest and provide `cloudbuild.yaml` out-of-the-box for
        building and pushing images unless local setup is explicitly requested.
    -   When building images with Cloud Build in the background you MUST provide
        the link where the user can monitor the long-running operation.

## Launching Apache Beam Pipelines with Dataflow Runner

-   When launching Python Pipelines without a Flex Template with
    `DataflowRunner`, you MUST scan the pipeline project directory for the
    following files:

    -   **`requirements.txt`**:
        -   If found, you MUST include `--requirements_file` pipeline option.
    -   **`setup.py`**:
        -   If found, you MUST include `--setup_file` pipeline option. This is
            critical if the pipeline uses local modules or packages.

-   When launching Python Pipelines with a Flex Template, if the Flex Template
    image is also the SDK Container image (Single Docker Image Configuration),
    then you MUST supply the image in the `sdk_container_image` parameter.

-   Confirm the launch command with the user.

### Lookup environment resources instead of using placeholder values

-   Avoid using generic placeholders (e.g., `your-gcp-project-id`) for GCP
    resources when drafting run scripts or configs. **Action**: If values are
    unknown, proactively run commands like `gcloud config get-value project` to
    find active resources to pre-fill scripts for the user. Confirm the values
    with the user before proceeding.

## Diagnostics & Troubleshooting

YOU MUST use this section when the user asks about performance of their dataflow
pipelines. This can be used to debug issues like pipeline slowness, pipeline
failures, etc.

### Task Execution Workflow

1.  **Understand User Request**: Extract Job ID, Project ID, Transform Name
    (optional), and Time Window.
2.  **Transform Name Mapping**: If the user requires transform-based debugging,
    map user-provided Transform Names to actual Dataflow `stage` or `ptransform`
    and apply to filters while querying:

    This mapping can be extracted from `gcloud dataflow jobs describe JOB_ID
    --full --format="json(pipelineDescription.executionPipelineStage)"`.

    1.  **Extract the targets**:
        *   Get stage_id: **`name`** property at the parent stage level. This
            matches `"F[digit]"` (e.g. `"F6"`).
        *   Get ptransform: inside the `componentTransform` array, read
            precisely from **`userName`** or **`originalTransform`** (e.g.
            `"RateLimitAndLog/ParMultiDo(RateLimitAndLog)"`). and use it as
            **`ptransform`**.
    2.  **Apply the filters strictly following mapping mechanics**:
        *   **For Cloud Logging queries**: Apply extracted ptransform name to
            filter `resource.labels.step_id="[Extracted ptransform name]"`.
        *   **For Monitoring queries**: Use the stage_id/ptransform filters
            based on filters supported by metric:
            `metric.labels.ptransform="[Extracted ptransform name]"` or
            `metric.labels.stage="[Extracted stage_id]"`.

3.  **Query Telemetry**:

    *   Use Dataflow REST API to get High level Job Messages/Events that
        happened in the job.
    *   Refer to
        [dataflow_diagnostics_reference.md](references/dataflow_diagnostics_reference.md)
        for key metrics and logging query patterns based on Job Type.
    *   Use Monitoring REST API to fetch metrics.
    *   Use GCloud Logging command to fetch logs.
    *   Use Dataflow REST API to fetch current snapshot metrics when historical
        time-series are not needed.

4.  **Analysis**:

    *   Correlate metrics spikes/drops with log errors.
    *   Identify Issues.

5.  **Output**: Provide a synthesized summary with symptoms, potential root
    cause, and links to relevant code transforms (using `file:///...` format).
    Follow this template to structure your response:

    1.  High level Job Events: Infer from job messages.
    2.  Data Freshness: Infer from watermark_age/system_lag metrics.
    3.  Throughput: Infer from
        elements_produced_count/estimated_bytes_produced_count metrics.
    4.  Backlog: Infer from estimated_backlog_processing_time/backlog_bytes
        metrics.
    5.  Bottlenecks: Infer from is_bottleneck/backlogged_keys metrics.
    6.  Autoscaling: Infer from horizontal_worker_scaling metric.
    7.  Recommendations: Provide recommendations based on the analysis of both
        metrics and logs.