

def run__NativeWrite(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.WRITE, transform_node.full_label, transform_node)
    if (transform.sink.format == 'text'):
        step.add_property(PropertyNames.FILE_NAME_PREFIX, transform.sink.file_name_prefix, with_type=True)
        step.add_property(PropertyNames.FILE_NAME_SUFFIX, transform.sink.file_name_suffix, with_type=True)
        step.add_property(PropertyNames.SHARD_NAME_TEMPLATE, transform.sink.shard_name_template, with_type=True)
        if (transform.sink.num_shards > 0):
            step.add_property(PropertyNames.NUM_SHARDS, transform.sink.num_shards, with_type=True)
        step.add_property(PropertyNames.VALIDATE_SINK, False, with_type=True)
    elif (transform.sink.format == 'bigquery'):
        step.add_property(PropertyNames.BIGQUERY_DATASET, transform.sink.table_reference.datasetId)
        step.add_property(PropertyNames.BIGQUERY_TABLE, transform.sink.table_reference.tableId)
        if (transform.sink.table_reference.projectId is not None):
            step.add_property(PropertyNames.BIGQUERY_PROJECT, transform.sink.table_reference.projectId)
        step.add_property(PropertyNames.BIGQUERY_CREATE_DISPOSITION, transform.sink.create_disposition)
        step.add_property(PropertyNames.BIGQUERY_WRITE_DISPOSITION, transform.sink.write_disposition)
        if (transform.sink.table_schema is not None):
            step.add_property(PropertyNames.BIGQUERY_SCHEMA, transform.sink.schema_as_json())
    elif (transform.sink.format == 'pubsub'):
        standard_options = transform_node.inputs[0].pipeline.options.view_as(StandardOptions)
        if (not standard_options.streaming):
            raise ValueError('PubSubSink is currently available for use only in streaming pipelines.')
        step.add_property(PropertyNames.PUBSUB_TOPIC, transform.sink.topic)
    else:
        raise ValueError(('Sink %r has unexpected format %s.' % (transform.sink, transform.sink.format)))
    step.add_property(PropertyNames.FORMAT, transform.sink.format)
    step.encoding = self._get_cloud_encoding(transform.sink.coder)
    step.add_property(PropertyNames.ENCODING, step.encoding)
    step.add_property(PropertyNames.PARALLEL_INPUT, {
        '@type': 'OutputReference',
        PropertyNames.STEP_NAME: input_step.proto.name,
        PropertyNames.OUTPUT_NAME: input_step.get_output(input_tag),
    })

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