

def _add_step(self, step_kind, step_label, transform_node, side_tags=()):
    'Creates a Step object and adds it to the cache.'
    from google.cloud.dataflow.internal import apiclient
    step = apiclient.Step(step_kind, self._get_unique_step_name())
    self.job.proto.steps.append(step.proto)
    step.add_property(PropertyNames.USER_NAME, step_label)
    self._cache.cache_output(transform_node, None, step_label)
    for tag in side_tags:
        self._cache.cache_output(transform_node, tag, step)
    return step

dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/runners/dataflow_runner.py DataflowPipelineRunner._add_step/VarMisuse@85/134 `step`->`step_label`