在使用Tensorflow Extended时，如何使用本地CSV-File运行apache beam管道？

pip install tensorflow-io tensorflow-transform apache-beam

import tensorflow as tf
import tensorflow_transform as tft
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from tensorflow_transform.beam import impl as beam_impl
from tensorflow_transform.tf_metadata import dataset_metadata
from tensorflow_transform.tf_metadata import dataset_schema

raw_data_metadata = dataset_metadata.DatasetMetadata(
    dataset_schema.from_feature_spec({
        'feature1': tf.io.FixedLenFeature([], tf.float32),
        'feature2': tf.io.FixedLenFeature([], tf.int64),
        'label': tf.io.FixedLenFeature([], tf.int64),
    })
)

pipeline_options = PipelineOptions()
with beam.Pipeline(options=pipeline_options) as pipeline:
    csv_data = (
        pipeline
        | 'ReadFromCSV' >> beam.io.ReadFromText('path/to/csv/file.csv')
    )

def parse_csv(row):
    columns = row.split(',')
    feature1 = float(columns[0])
    feature2 = int(columns[1])
    label = int(columns[2])
    return {
        'feature1': feature1,
        'feature2': feature2,
        'label': label,
    }

parsed_data = csv_data | 'ParseCSV' >> beam.Map(parse_csv)

def preprocessing_fn(inputs):
    feature1_scaled = inputs['feature1'] / tf.reduce_max(inputs['feature1'])
    feature2_scaled = inputs['feature2'] / tf.reduce_max(inputs['feature2'])
    return {
        'feature1_scaled': feature1_scaled,
        'feature2_scaled': feature2_scaled,
        'label': inputs['label'],
    }

transformed_data, transform_fn = (
    (parsed_data, raw_data_metadata)
    | 'AnalyzeAndTransform' >> beam_impl.AnalyzeAndTransformDataset(preprocessing_fn)
)

(transformed_data[0]
    | 'EncodeTFRecord' >> beam.Map(tf.io.encode_proto_as_string)
    | 'WriteTFRecord' >> beam.io.WriteToTFRecord('path/to/output.tfrecord')
)