我正在尝试创建一个管道，该管道正在从Pubsub读取一些数据并写入postgres数据库。

pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(StandardOptions).streaming = True
Device = NamedTuple(
     "Device",
     [
         ("id", str),
         ("userId", str),
         ("patientId", str)
     ])
coders.registry.register_coder(Device, coders.RowCoder)
p = beam.Pipeline(options = pipeline_options)

(p
   | 'ReadFromPubSub' >> beam.io.gcp.pubsub.ReadFromPubSub(topic="projects/us-vpc/topics/pipeline").with_output_types(bytes)
   | "Decode" >> beam.Map(lambda x: x.decode('utf-8'))
   | beam.Map(lambda x:
             Device(id= "e4f63782-66f5-4f49-911f-0b00efa5b23e", userId="Random",
                             patientId=str('12345')))
     .with_output_types(Device)
   | beam.WindowInto(beam.window.FixedWindows(1))
   | 'Write to jdbc' >> WriteToJdbc(
       table_name= "patient_device",
       driver_class_name = db_driver_name,
       jdbc_url = jdbc:postgresql://localhost:5432/db_name,
       username = dev-user,
       password = db-password
       )
) 
result = p.run()
result.wait_until_finish()

我可以看到在数据流部署到gcp后正在创建四个步骤。

1. 从Pub Sub阅读
2. 解码
3. 地图
4. 窗口输入

但是问题是“写入jdbc”步骤没有在数据流上创建。

这是执行数据流的命令：

python pipeline.py --runner DataflowRunner --project us-con-project-location --temp_location gs://staging/temp --staging_location gs://staging/temp --region us-east1 --input_topic "projects/us-vpc/topics/pipeline" --subnetwork regions/us-east1/subnetworks/-public-01

任何帮助将不胜感激！

例子如下：https://beam.apache.org/releases/pydoc/2.24.0/apache_beam.io.jdbc.html

带有JdbcIO的Apache Beam管道