Dataflow Pipeline Development Skill

This skill guides the agent through developing, modifying, testing, linting, packaging, and building Apache Beam streaming pipelines in Python and Java for Google Cloud Dataflow.

---

1. Python Pipeline Development Workflow

Python pipelines are located in pipelines/<use_case>/.

Step 1: Environment & Dependencies

Ensure Python 3.13+ is used. In the pipeline directory:

# Create and activate virtual environment
python3 -m venv .venv
source .venv/bin/activate

# Install requirements
pip install -r requirements.txt
if [ -f "requirements-dev.txt" ]; then
  pip install -r requirements-dev.txt
fi

Step 2: Code Formatting (Google Style)

Run yapf across the pipeline directory:

yapf -i -r --style yapf .

Step 3: Linting with Google PyLint Configuration

Lint the code against the shared root pipelines/pylintrc config:

pylint --rcfile ../pylintrc .

Fix all lint errors (e.g. docstrings, naming conventions, import ordering).

Step 4: Unit Testing with Pytest

Run unit tests across test suites with pytest:

pytest tests/ -v

Step 5: Package Validation

Verify that the setup.py packages all sub-modules correctly for Dataflow workers:

python setup.py sdist

Step 6: Local Execution with DirectRunner

Test pipeline execution locally before submitting to the cloud:

python main.py \
  --runner=DirectRunner \
  --project=test-project \
  --temp_location=/tmp/dataflow-temp

Step 7: Custom SDK Container Build (if required)

For pipelines using GPU acceleration, custom C/Python libraries, or specialized base images (e.g. ml_ai_python, anomaly_detection, cdp, iot_analytics, marketing_intelligence):

• SDK Version Parity: Verify that the apache/beam_python3.13_sdk:<version> tag in Dockerfile matches requirements.txt (apache-beam[gcp]==<version>).

# Set required image tag
export CONTAINER_URI="gcr.io/$PROJECT/dataflow-ml-custom:latest"

# Build and push using Cloud Build
gcloud builds submit \
  --region=$REGION \
  --default-buckets-behavior=regional-user-owned-bucket \
  --substitutions _TAG=$CONTAINER_URI \
  .

---

2. Java Pipeline Development Workflow

Java pipelines are located in pipelines/<use_case>_java/.

Step 1: Gradle Build & Test

Verify code compilation and test execution:

./gradlew build

Step 2: Code Formatting (Spotless)

Apply Google Java Style automatically:

./gradlew spotlessApply

Step 3: Local Testing with DirectRunner

Run the pipeline locally:

./gradlew run -Pargs="--runner=DirectRunner --project=test-project --gcpTempLocation=/tmp/dataflow-temp"

---

3. Apache Beam Best Practices in this Repo

1. SDK & Custom Container Parity: The Apache Beam SDK version in the pipeline code (requirements.txt) and worker container (Dockerfile) must be identical. Version divergence leads to serialization failures, container harness mismatch, and worker initialization crashes on Dataflow.
2. Option Classes: Define pipeline arguments by subclassing PipelineOptions (Python) or PipelineOptions interface (Java) in options.py / options/.
3. Worker Isolation: Always enforce private IP communication:
   • Python: --no_use_public_ip
   • Java: --usePublicIps=false
4. Dead-Letter Queues (Side Outputs): For unparseable or error records, route failed elements to a dead-letter output or BigQuery error table rather than crashing worker threads:

   # Python side-output pattern
   parsed_records, errors = (
       raw_records
       | 'ParseRecords' >> beam.ParDo(ParseDoFn()).with_outputs('errors', main='valid')
   )
   

5. Metrics Instrumentation: Use Beam metrics to track throughput and error counts:

   self.processed_counter = Metrics.counter(self.__class__, 'processed_elements')
   self.processed_counter.inc()