skills.rest
ricardolui
ricardoluiCOMMUNITY
Explore all skills by this author →
GitHub

blip-composer-dataform-stream-migration

Migrates Cloud Composer and Dataform streaming pipelines from Dataflow jobs to Confluent Kafka Iceberg ingestion.

Updated Jul 7, 2026
Search Tags:#migration#kafka#bigquery#iceberg#dataflow#dataform#cloud composer
One-click install
npx skills add https://github.com/ricardolui/gcp-custom-agent-skills --skill blip-composer-dataform-stream-migration-ricardolui
Or copy as Structured Prompt for Agent▼
Please help me install this Agent Skill.
Skill: blip-composer-dataform-stream-migration
Source: https://github.com/ricardolui/gcp-custom-agent-skills/tree/main/blip-composer-dataform-stream-migration
Command: npx skills add https://github.com/ricardolui/gcp-custom-agent-skills --skill blip-composer-dataform-stream-migration-ricardolui

SYSTEM DOCUMENTATION & REQUIREMENTS

💡 This Skill includes references (resource) components.

What problem does it solve? Legacy Cloud Composer DAGs run redundant Dataflow streaming jobs alongside Dataform, duplicating BigQuery Storage Write API ingestion costs and wasting thousands of dollars per month in GCE compute. This Skill guides the elimination of Dataflow operators and the cutover to Confluent Kafka Connect and Pub/Sub Import streaming directly into BigLake Iceberg tables. ## Core Features & Use Cases - DAG Refactoring: Strip Dataflow operators from Composer DAGs and rebuild them as pure incremental Dataform compilation and invocation routines with parameterized source datasets. - Shadow A/B Validation: Provision isolated __kfkconn_test datasets and run full outer join parity audits comparing production and shadow tables before cutover. - Source Mapping Resolution: Create BigQuery compatibility views when Confluent SMT prefixes table names, preventing Table not found errors in Dataform declarations. - Use Case: A data engineer needs to migrate 58 Kafka ingestion DAGs off Dataflow. They follow the 6-phase lifecycle: align sources, parameterize Dataform, create shadow datasets, refactor the DAG, validate parity, then drain the legacy Dataflow jobs. ## Quick Start Ask the assistant to refactor a specific Composer ingestion DAG to remove its Dataflow operators and generate a pure incremental Dataform DAG with shadow test configuration for A/B parity validation.

Frequently Asked Questions about blip-composer-dataform-stream-migration

High-intent search queries and answers about installing and using this skill.

FAQPage Schema
How do I remove Dataflow jobs from Cloud Composer DAGs?▼

Strip all Dataflow operators such as DataflowStartFlexTemplateOperator and status-check branches, then retain only DataformCreateCompilationResultOperator and DataformCreateWorkflowInvocationOperator. This converts the DAG into a pure incremental Dataform routine running in under 30 seconds.

How to validate data parity between production and shadow BigQuery tables?▼

Run a full outer join query comparing row counts and distinct primary keys per partition date between the production silver table and the `__kfkconn_test` shadow table. Closed past dates must show zero row difference for exact parity before cutover.

Why does Dataform fail with Table raw_platform_kfkconn not found?▼

Confluent Kafka Connect SMT prefixes topic names, so tables may be named `imt_prod_<name>` while Dataform expects `imt_<name>`. Create a BigQuery compatibility view in the kfkconn dataset pointing to the actual Iceberg table to fix it without code changes.

Can I upload DAGs to the Composer bucket with my user account?▼

Direct user identities typically lack write access to the Composer GCS bucket and receive 403 errors. Use gcloud storage cp with the --impersonate-service-account flag targeting the Composer management service account.

When should I not use this Dataflow elimination approach?▼

Do not use it for batch database ingestion such as SQL Server or MongoDB jobs that still require Dataflow or Dataproc. It also does not cover provisioning new Confluent connectors or migrating Azure Event Hubs via Pub/Sub Import, which are handled by separate workflows.

Related Skills

Explore More Data & Analytics Skills→
cardox6cardox6

neo4j-kafka-skill

Configure Neo4j Kafka sink and source connectors and the native CDC API.

Community
Advanced
reason-machinesreason-machines
5

google-cloud-data-engineering-hub

Build GCP data pipelines using BigQuery, Dataflow, Beam, Composer, Pub/Sub, and Vertex AI.

Official
Advanced
Sir-chawakornSir-chawakorn
9

diff-table-parity

Compare two datasets for row counts, key differences, and column value mismatches.

Community
Advanced
ajurcenkajurcenk
1

redpanda-migrator

Generate Redpanda Migrator YAML configurations for Kafka-compatible cluster migrations.

Community
Advanced