

def run_CombineValues(self, transform_node):
    transform = transform_node.transform
    input_tag = transform_node.inputs[0].tag
    input_step = self._cache.get_pvalue(transform_node.inputs[0])
    step = self._add_step(TransformNames.COMBINE, transform_node.full_label, transform_node)
    fn_data = (transform.fn, transform.args, transform.kwargs, ())
    step.add_property(PropertyNames.SERIALIZED_FN, pickler.dumps(fn_data))
    step.add_property(PropertyNames.PARALLEL_INPUT, {
        '@type': 'OutputReference',
        PropertyNames.STEP_NAME: input_step.proto.name,
        PropertyNames.OUTPUT_NAME: input_step.get_output(input_tag),
    })
    accumulator_encoding = self._get_typehint_based_encoding(self._get_transform_type_hint(transform_node), window_value=False)
    output_encoding = self._get_typehint_based_encoding(self._get_transform_type_hint(transform_node))
    step.encoding = output_encoding
    step.add_property(PropertyNames.ENCODING, accumulator_encoding)
    outputs = []
    outputs.append({
        PropertyNames.USER_NAME: ('%s.%s' % (transform_node.full_label, PropertyNames.OUT)),
        PropertyNames.ENCODING: step.encoding,
        PropertyNames.OUTPUT_NAME: PropertyNames.OUT,
    })
    step.add_property(PropertyNames.OUTPUT_INFO, outputs)

dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/runners/dataflow_runner.py DataflowPipelineRunner.run_CombineValues/original