Runtime Parameters

1. Introduction

2. Why This Concept Exists

• Compile-time evaluation: If you write beam.io.ReadFromText(my_path) where my_path is a simple string, that value is locked into the graph structure.
• The Issue: If you build a reusable template, you cannot change the input path without recompiling and redeploying the template.

Runtime Parameters solve this limitation. By delaying parameter evaluation, templates can be built once and run repeatedly with different options, allowing orchestrators to trigger the pipeline dynamically.

3. Key Terminology

• ValueProvider: A container class in the Apache Beam SDK that wraps a variable, delaying its resolution until the pipeline actually starts executing on workers.
• Static Value: A standard python variable resolved immediately during code compilation.
• Runtime Value: A value supplied only when a job is launched, mapped to a ValueProvider object.
• NestedValueProvider: A helper class used to transform or combine the outputs of one or more ValueProvider objects during execution.

4. How It Works

1. Graph Construction: The SDK creates placeholders for these variables. Since their values are unknown, the code must treat them as wrapper objects rather than raw strings or integers.
2. Job Staging: The job graph containing these placeholders is submitted.
3. Job Launch: The user launches the template and supplies values (e.g. --input_file=gs://bucket/data.csv).
4. Worker Resolution: When worker VMs boot, the Dataflow runner injects the values into the ValueProvider placeholders. Within custom transforms (like DoFn processes), you retrieve the raw value by calling .get() on the wrapper.

5. Visual Diagram

6. Code Example

pythonCopy
import logging
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# 1. Define custom options using ValueProvider
class PipelineRuntimeOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument(
            "--cutoff_limit",
            type=int,
            default=50,
            help="Cutoff integer threshold for filtering"
        )

# 2. Access the value inside DoFn
class FilterByThreshold(beam.DoFn):
    def __init__(self, threshold_provider):
        # Store the provider reference (do not call .get() here!)
        self.threshold_provider = threshold_provider

    def process(self, element):
        # Retrieve the value at runtime on the workers
        limit = self.threshold_provider.get()
        if element > limit:
            yield element

def run():
    options = PipelineOptions()
    runtime_opts = options.view_as(PipelineRuntimeOptions)
    
    with beam.Pipeline(options=options) as p:
        (p
         | "CreateNumbers" >> beam.Create([10, 45, 60, 80, 20])
         | "FilterNumbers" >> beam.ParDo(FilterByThreshold(runtime_opts.cutoff_limit))
         | "Log" >> beam.Map(print)
        )

if __name__ == "__main__":
    logging.getLogger().setLevel(logging.INFO)
    run()

7. Code Explanation

• parser.add_value_provider_argument registers the parameter. The option returned (runtime_opts.cutoff_limit) is an instance of ValueProvider, not an int.
• Inside FilterByThreshold.__init__, we accept the ValueProvider object and save it as a class attribute.
• CRITICAL: We must not call self.threshold_provider.get() inside __init__. The __init__ constructor runs during graph compilation on the launch machine where the value is still unset, which would trigger a crash.
• Inside process, which runs on the worker VMs during record processing, calling self.threshold_provider.get() safely retrieves the runtime value.

8. Real Production Example

9. Common Mistakes

• Calling .get() during DAG construction: Attempting to use .get() directly in the pipeline setup code. E.g., cutoff = runtime_opts.cutoff_limit.get(). This will crash with a RuntimeError stating that the value is not available.
• Using Non-Supported Transforms: Passing ValueProvider objects to transforms that do not explicitly accept them. For example, some custom third-party SDK connectors do not accept ValueProviders for configuration fields.

10. Best Practices

• Always implement fallback values in add_value_provider_argument using the default parameter.
• Wrap complex transformations on runtime values using NestedValueProvider to avoid messy worker-side conversions.
• Verify that your sinks and sources support ValueProvider arguments in the version of Apache Beam you are using.

11. Summary

• Runtime parameters decouple pipeline code from dynamic variables.
• ValueProvider objects defer variable resolution until execution.
• Never call .get() during graph initialization; only call it inside worker-side execution logic (e.g. process).
• Templates use runtime parameters to run dynamic workloads.

12. Interactive Challenges

13. Related Content

• Flex Templates
• Pipeline Options
• Introduction to Dataflow