class klio.transforms.helpers.KlioMessageCounter

Helper transform to count elements.

This transform will yield elements given to it, counting each one it sees.

Since it is a DoFn transform, it needs to be invoked via beam.ParDo when used. Some usage examples:

# Example composite transform
class MyCompositeTransform(beam.PTransform):
    def __init__(self, *args, **kwargs):
        self.transform = MyTransform(*args, **kwargs)
        self.ctr = KlioMessageCounter(suffix="processed", "MyTransform")

    def expand(self, pcoll):
        return (
            | "Process items" >> self.transform
            | "Count items processed" >> beam.ParDo(self.ctr)

# Example pipeline segment
def run(input_pcol, config):
    input_data = input_pcol | helpers.KlioGcsCheckInputExists()
    data_not_found = input_data.not_found | helpers.KlioMessageCounter(
  • suffix (str) – suffix of the counter name. The full counter name will be kmsg-{suffix}.

  • bind_transform (str) – Name of transform to bind the counter to. This is used for Dataflow monitoring UI purposes, and can be set to a prior or following transform in a pipeline, or to itself.

class klio.transforms.helpers.KlioGcsCheckInputExists

Klio transform to check input exists in GCS.

class klio.transforms.helpers.KlioGcsCheckOutputExists

Klio transform to check output exists in GCS.

class klio.transforms.helpers.KlioFilterPing

Klio transform to tag outputs if in ping mode or not.

class klio.transforms.helpers.KlioFilterForce

Klio transform to tag outputs if in force mode or not.

class klio.transforms.helpers.KlioWriteToEventOutput

Klio composite transform to write to the configured event output.

class klio.transforms.helpers.KlioDrop

Klio DoFn to log & drop a KlioMessage.

class klio.transforms.helpers.KlioCheckRecipients

Check if current job should handle a received v2 message.

class klio.transforms.helpers.KlioUpdateAuditLog

Update a KlioMessage’s audit log to include current job.

class klio.transforms.helpers.KlioDebugMessage

Log KlioMessage.

  • prefix (str) – logging prefix. Default: "DEBUG".

  • log_level (str or int) – The desired log level for the KlioMessage logs. See available log levels for what’s supported. Default: "INFO".

class klio.transforms.helpers.KlioSetTrace

Insert a Python debugger trace point.

class klio.transforms.helpers.KlioTriggerUpstream

Trigger upstream job from current job with a given KlioMessage.

This transform will update the intended recipients in KlioMessage. metadata in order to trigger a partial bottom-up execution of the overall graph of jobs. It will also generate a log message (optional), then publish the KlioMessage to the upstream’s Pub/Sub topic.


Klio does not automatically trigger upstream jobs if input data does not exist. It must be used manually within a job’s pipeline definition (in


In order to get access to input data not found, the automatic data existence check that Klio does must be turned off by setting[].skip_klio_existence_check to True. Then the existence check must be invoked manually. See example and klio-job.yaml files below.

Example usage:

import apache_beam as beam
from klio.transforms import helpers
import transforms

def run(input_pcol, config):
    # use the default helper transform to do the default input check
    # in order to access the output tagged with `not_found`
    input_data = input_pcol | helpers.KlioGcsCheckInputExists()

    # Pipe the input data that was not found (using Tagged Outputs)
    # into `KlioTriggerUpstream` in order to update the KlioMessage
    # metadata, log it, then publish to upstream's
    _ = input_data.not_found | helpers.KlioTriggerUpstream(

    # pipe the found input pcollection into transform(s) as needed
    output_pcol = input_data.found | beam.ParDo(MyTransform())
    return output_pcol
# Example klio-job.yaml
version: 2
job_name: my-job
  project: my-gcp-project
  # `KlioTriggerUpstream` only supports streaming jobs
  streaming: True
  # <-- snip -->
      - type: pubsub
        topic: projects/my-gcp-project/topics/upstream-topic-output
        subscription: projects/my-gcp-project/subscriptions/my-job-in
    # <-- snip -->
      - type: gcs
        location: gs://my-gcp-project/upstream-output-data
        file_suffix: .ogg
        # Be sure to skip Klio's default input existence check in
        # order to access the input data that was not found.
        skip_klio_existence_check: True
  • upstream_job_name (str) – Name of upstream job.

  • upstream_topic (str) – Pub/Sub topic for the upstream job, in the form of project/<GCP_PROJECT>/topics/<TOPIC_NAME>.

  • log_level (str, int, or None) –

    The desired log level for log message, or None if no logging is desired. See available log levels for what’s supported. Default: "INFO".


SystemExit – If the current job is not in streaming mode (set in klio-job.yaml::pipeline_options.streaming), if the provided log level is not recognized, or if the provided upstream topic is not in the correct form.