Bigtable Column Family Completeness Gate with Kestra

id: bigtable-column-family-completeness-gate
namespace: company.team

description: |
  Sample rows from a Bigtable table, classify compliance as COMPLIANT, WARNING, CRITICAL or
  PROBE_FAILED based on how many sampled rows are missing a required column family, alert
  Slack with the exact row keys and missing families, and log a final audit summary either way.

inputs:
  - id: instance_id
    type: STRING
    displayName: Bigtable instance ID
    description: The Bigtable instance ID. Maps to the ReadRows task's instanceId field.
    defaults: "demo-instance"

  - id: table_id
    type: STRING
    displayName: Bigtable table ID
    description: The table to audit. Maps to the ReadRows task's tableId field.
    defaults: "sensor-readings"

  - id: row_key_prefix
    type: STRING
    displayName: Row key prefix
    description: Row key prefix scoping the audit, for example device- to only audit
      one tenant or device namespace. Leave empty to scan the whole table.
    defaults: ""

  - id: required_column_families
    type: JSON
    displayName: Required column families
    description: Every sampled row is expected to contain at least one cell from
      each of these column families.
    defaults: '["readings", "metadata"]'

  - id: column_family_check_mode
    type: SELECT
    displayName: Column family check mode
    description: any_required_missing flags a row missing even one required family,
      strict schema compliance. all_required_missing only flags a row missing
      every required family, catching rows that are effectively empty rather
      than partially written.
    values:
      - "any_required_missing"
      - "all_required_missing"
    defaults: "any_required_missing"

  - id: sample_size
    type: INT
    displayName: Sample size
    description: Maximum number of rows read per audit pass. Maps to the ReadRows
      task's limit field.
    defaults: 100

  - id: violation_count_critical_threshold
    type: INT
    displayName: Critical violation count threshold
    description: Violation count at or above which the run is classified CRITICAL
      instead of WARNING.
    defaults: 10

  - id: use_emulator
    type: BOOL
    displayName: Use local Bigtable emulator
    description: When true, connects to a local Bigtable emulator (emulator_host)
      with no credentials, for local testing. When false, connects to real
      Bigtable with GCP_SERVICE_ACCOUNT and GCP_PROJECT_ID.
    defaults: true

  - id: emulator_host
    type: STRING
    displayName: Emulator host
    description: host:port of the local Bigtable emulator, only used when
      use_emulator is true.
    defaults: "localhost:8086"

tasks:
  - id: route_connection_mode
    type: io.kestra.plugin.core.flow.If
    description: Picks the connection. The emulator branch uses emulatorHost with no
      credentials, the live branch uses a real GCP service account.
    condition: "{{ inputs.use_emulator }}"
    then:
      - id: probe_rows_emulator
        type: io.kestra.plugin.gcp.bigtable.ReadRows
        description: Read-only. Samples up to sample_size rows under row_key_prefix from
          the local emulator.
        allowFailure: true
        projectId: "emulator-project"
        instanceId: "{{ inputs.instance_id }}"
        emulatorHost: "{{ inputs.emulator_host }}"
        tableId: "{{ inputs.table_id }}"
        rowKeyPrefix: "{{ inputs.row_key_prefix }}"
        limit: "{{ inputs.sample_size }}"
        fetchType: FETCH
    else:
      - id: probe_rows_live
        type: io.kestra.plugin.gcp.bigtable.ReadRows
        description: Read-only. Samples up to sample_size rows under row_key_prefix from
          real Bigtable.
        allowFailure: true
        projectId: "{{ secret('GCP_PROJECT_ID') }}"
        serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"
        instanceId: "{{ inputs.instance_id }}"
        tableId: "{{ inputs.table_id }}"
        rowKeyPrefix: "{{ inputs.row_key_prefix }}"
        limit: "{{ inputs.sample_size }}"
        fetchType: FETCH

  - id: audit_column_families
    type: io.kestra.plugin.core.output.OutputValues
    description: Flags every sampled row missing a required column family, per
      column_family_check_mode. Empty when neither probe branch returned.
    values:
      violations: "{{ (outputs.probe_rows_emulator is not defined and
        outputs.probe_rows_live is not defined) ? '[]' : ({'required':
        inputs.required_column_families, 'rows': (outputs.probe_rows_emulator is
        defined ? outputs.probe_rows_emulator.rows :
        outputs.probe_rows_live.rows) ?? []} |
        jq(inputs.column_family_check_mode == 'all_required_missing' ?
        '.required as $req | [((.rows // [])[]) | . as $r | (($r.cells // []) |
        map(.family) | unique) as $present | ($req - $present) as $missing |
        select(($missing|length) == ($req|length)) | {rowKey: $r.rowKey,
        missing_families: $missing}]' : '.required as $req | [((.rows // [])[])
        | . as $r | (($r.cells // []) | map(.family) | unique) as $present |
        ($req - $present) as $missing | select(($missing|length) > 0) | {rowKey:
        $r.rowKey, missing_families: $missing}]') | first | toJson) }}"

  - id: classify_compliance_status
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when neither probe branch returned, CRITICAL or
      WARNING when sampled rows are missing required families past or under
      threshold, COMPLIANT when none are.
    format: "{{ (outputs.probe_rows_emulator is not defined and
      outputs.probe_rows_live is not defined) ? 'PROBE_FAILED' :
      ((fromJson(outputs.audit_column_families.values.violations) | length) >=
      inputs.violation_count_critical_threshold ? 'CRITICAL' :
      ((fromJson(outputs.audit_column_families.values.violations) | length) > 0
      ? 'WARNING' : 'COMPLIANT')) }}"

  - id: route_by_compliance_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a compliant sample. Branch B alerts Slack when
      sampled rows are missing a required column family, or the probe itself
      failed.
    value: "{{ outputs.classify_compliance_status.value }}"
    cases:
      "COMPLIANT":
        - id: log_compliant
          type: io.kestra.plugin.core.log.Log
          message: "Bigtable {{ inputs.instance_id }}/{{ inputs.table_id }} compliant.
            Every sampled row under prefix '{{ inputs.row_key_prefix }}' has the
            required column families."
      "WARNING":
        - id: alert_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "Bigtable " ~ inputs.instance_id ~ "/" ~ inputs.table_id ~ " WARNING. " ~ (fromJson(outputs.audit_column_families.values.violations) | length) ~ " sampled row(s) missing a required column family: " ~ (fromJson(outputs.audit_column_families.values.violations) | jq('map(.rowKey + ": missing " + (.missing_families | join(", "))) | join("; ")') | first) ~ "."} | toJson }}
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "Bigtable " ~ inputs.instance_id ~ "/" ~ inputs.table_id ~ " CRITICAL. " ~ (fromJson(outputs.audit_column_families.values.violations) | length) ~ " sampled row(s) missing a required column family, at or above the " ~ inputs.violation_count_critical_threshold ~ " threshold. Sample: " ~ (fromJson(outputs.audit_column_families.values.violations) | jq('.[0:5] | map(.rowKey + ": missing " + (.missing_families | join(", "))) | join("; ")') | first) ~ "."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "Bigtable column family audit on " ~ inputs.instance_id ~ "/" ~ inputs.table_id ~ " produced no result. Check instance_id, table_id, and either emulator_host or GCP_SERVICE_ACCOUNT depending on use_emulator."} | toJson }}

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "Bigtable column family completeness gate completed. table={{
      inputs.instance_id }}/{{ inputs.table_id }} status={{
      outputs.classify_compliance_status.value }} mode={{
      inputs.column_family_check_mode }} use_emulator={{ inputs.use_emulator }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {"text": "Bigtable column family completeness gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always against the local emulator by default.
    cron: "0 * * * *"
    disabled: true
    inputs:
      instance_id: "demo-instance"
      table_id: "sensor-readings"
      row_key_prefix: ""
      required_column_families: '["readings", "metadata"]'
      column_family_check_mode: "any_required_missing"
      sample_size: 100
      violation_count_critical_threshold: 10
      use_emulator: true
      emulator_host: "localhost:8086"

Bigtable has no schema to enforce, a column family exists the moment any cell is written to it, and nothing stops a buggy write path, a half-finished migration, or a dropped field in an application update from silently producing rows that are missing a family every downstream reader assumes is there. The table itself will never tell you, a read for a missing family just comes back empty, no error. This blueprint turns a row sample into a Day 2 compliance gate: it reads a batch of real rows, checks each one against the column families it is supposed to have, and alerts Slack with the exact row keys and families missing, never a guess.

How it works

route_connection_mode (io.kestra.plugin.core.flow.If) picks the connection based on use_emulator. probe_rows_emulator connects to a local Bigtable emulator via emulatorHost with no credentials, probe_rows_live connects to real Bigtable with a GCP service account. Both run io.kestra.plugin.gcp.bigtable.ReadRows with rowKeyPrefix, limit: sample_size, and fetchType: FETCH, and both have allowFailure: true so an unreachable instance or missing table does not stop the flow before it can be classified as PROBE_FAILED below.
audit_column_families (io.kestra.plugin.core.output.OutputValues) reads whichever branch's rows list exists (guarded with is defined/is not defined, since exactly one of the two probe tasks ran), computes the distinct cells[].family values present on each row, and flags every row missing a required family per column_family_check_mode.
classify_compliance_status (io.kestra.plugin.core.debug.Return) returns PROBE_FAILED when neither probe produced output, CRITICAL or WARNING depending on whether the violation count meets violation_count_critical_threshold, or COMPLIANT when there are none.
route_by_compliance_status (io.kestra.plugin.core.flow.Switch) branches on that status: COMPLIANT logs it, WARNING, CRITICAL, and PROBE_FAILED each alert Slack with the exact row keys and missing families from the audit, never an invented count.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real answer to whether sampled rows actually have the shape your downstream readers assume, not a hope that every write path stayed in sync.
A Slack alert naming the exact row keys and the exact missing families, never an invented number.
A fully local test path: the default inputs point at a Bigtable emulator, so the whole gate can be exercised without a GCP account.

Who it's for

Data platform teams running Bigtable for time-series, IoT, or event data who need to catch a silently broken write path before a downstream job fails on a missing field.
Teams migrating or reshaping a Bigtable schema who want proof that every row in a namespace already has the new column family before cutting over readers to depend on it.
Teams evaluating the Bigtable plugin who want a working example of io.kestra.plugin.gcp.bigtable.ReadRows against both the local emulator and real Bigtable, using the same task with different connection properties.

Why orchestrate this with Kestra

Bigtable itself enforces nothing about which families a row contains, there is no constraint, no alert, no concept of "this row looks wrong". Kestra adds the recurring sample, the column family completeness check, and a Slack alert built from the rows' own data, all as one declarative YAML file you can version and review like code, runnable against a local emulator before it ever touches a real table.

Prerequisites

For local testing, the official Bigtable emulator, no GCP account needed:
```
docker run -p 127.0.0.1:8086:8086 --rm -ti google/cloud-sdk gcloud beta emulators bigtable start --host-port=0.0.0.0:8086
```
Port 8086 does not collide with Kestra's own default UI port. The emulator supports table creation and read/write against any project and instance name, it does not support the full instance/cluster administrative API, which this blueprint does not use.
For real Bigtable, a GCP project with a Bigtable instance and table, and a service account with bigtable.rows.readColumnFamilies-equivalent read permission (roles/bigtable.reader is enough, this blueprint only reads).
A Slack incoming webhook for alerts.
This plugin calls the Bigtable Data API directly from the Kestra worker process, there is no sub-container taskRunner. If Kestra itself runs in a container and use_emulator is true, point emulator_host at host.docker.internal:8086 or connect Kestra's container to the emulator's Docker network and use its container name instead.

Secrets

GCP_PROJECT_ID, GCP_SERVICE_ACCOUNT: used by probe_rows_live only, not needed when use_emulator is true.
SLACK_WEBHOOK_URL: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set all three as base64-encoded environment variables: SECRET_GCP_PROJECT_ID, SECRET_GCP_SERVICE_ACCOUNT, SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Type	Default	Allowed values
`instance_id`	STRING	`demo-instance`	any Bigtable instance ID
`table_id`	STRING	`sensor-readings`	any Bigtable table ID
`row_key_prefix`	STRING	`` (empty)	any row key prefix, empty scans the whole table
`required_column_families`	JSON	`["readings", "metadata"]`	JSON array of column family names
`column_family_check_mode`	SELECT	`any_required_missing`	`any_required_missing`, `all_required_missing`
`sample_size`	INT	`100`	any positive integer
`violation_count_critical_threshold`	INT	`10`	any non-negative integer
`use_emulator`	BOOL	`true`	`true`, `false`
`emulator_host`	STRING	`localhost:8086`	any `host:port`

Quick start

Start the Bigtable emulator, create a table and write a few rows with the cbt CLI or the Bigtable client of your choice, some with every required family, some without.
Add the SLACK_WEBHOOK secret.
Run this flow manually with the defaults and read log_audit_summary for the result.
Set use_emulator: false and add the GCP_PROJECT_ID and GCP_SERVICE_ACCOUNT secrets to point this at a real table.
Enable the hourly trigger.

How to extend

Narrow row_key_prefix to the namespace of a single tenant or device and run one instance of this flow per tenant with different required_column_families, if different tenants are expected to have different schemas.
Add a qualifier-level check inside a required family, not just family presence, for example requiring a specific column like metadata:schema_version to exist.
Chain a downstream task after CRITICAL that writes the exact violating row keys to a file in internal storage for a data engineer to inspect, instead of only Slack.

Pitfalls

sample_size is a sample, not a full table scan. A rare violation pattern can be missed entirely if it falls outside the sampled rows, raise sample_size or scope row_key_prefix tighter for a more complete audit.
A row with zero cells at all (never written, or fully deleted) has no families and is correctly flagged a violation under both column_family_check_mode settings, that is expected, not a false positive.
The Bigtable emulator does not persist data across container restarts by default and does not support the full administrative API, do not rely on it for anything beyond local testing of this gate's logic.
emulatorHost uses plaintext transport with no credentials, only ever point it at a local emulator, never at a real endpoint.

Bigtable Column Family Completeness Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Bigtable Column Family Completeness Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

BigQuery Partition-Aware Incremental Refresh

BigQuery High-Cost Query and Slot Contention Sentinel

Track Bitcoin Invoice Payments to Final Confirmation with walletnotify

Apache Beam Word Count with the DIRECT Runner

BigQuery Table Partition and Cost Governance Audit

Batch Bitcoin Payouts When Fees Are Low, with Human Approval

More Infrastructure Blueprints

Monitor FastAPI Synthetic Latency SLOs and Alert SRE Teams via Slack

Fastly Surrogate Key Purge for CMS Content Updates

Feature-Flag Progressive Delivery with an Error-Rate Guard

Faktory Job Enqueue Path Reachability Sentinel

Fastly Month-to-Date Cost Guard

Falco Cloud Native Runtime Threat and Security Priority Event Sentinel

New to Kestra?