def run_Create(self, transform_node):
    transform = transform_node.transform
    step = self._add_step(TransformNames.CREATE_PCOLLECTION,
                          transform_node.full_label, transform_node)
    # TODO(silviuc): Eventually use a coder based on typecoders.
    # Note that we base64-encode values here so that the service will accept
    # the values.
    element_coder = coders.PickleCoder()
    step.add_property(
        PropertyNames.ELEMENT,
        [base64.b64encode(element_coder.encode(v))
         for v in transform.value])
    # The service expects a WindowedValueCoder here, so we wrap the actual
    # encoding in a WindowedValueCoder.
    step.encoding = self._get_cloud_encoding(
        coders.WindowedValueCoder(element_coder))
    step.add_property(
        PropertyNames.OUTPUT_INFO,
        [{PropertyNames.USER_NAME: (
            '%s.%s' % (transform_node.full_label, PropertyNames.OUT)),
          PropertyNames.ENCODING: step.encoding,
          PropertyNames.OUTPUT_NAME: PropertyNames.OUT}])
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/runners/dataflow_runner.py/DataflowPipelineRunner.run_Create/original