如何在apache-beam python中持久化外部获取的有状态数据？

from apache_beam.transforms import DoFn
from apache_beam.transforms import ParDo
from apache_beam.transforms import window
from apache_beam.transforms.state import BagStateSpec
from apache_beam.transforms.state import ReadModifyWriteStateSpec

class MyDoFn(DoFn):
    def __init__(self):
        # 定义状态
        self.state = None

    def setup(self):
        # 初始化状态
        self.state = self.state_bag.read()

    def process(self, element, window=DoFn.WindowParam):
        # 处理数据并更新状态
        # ...

    def finish_bundle(self):
        # 在bundle结束时将状态写回
        self.state_bag.write(self.state)

with beam.Pipeline() as p:
    # 从外部获取数据集
    input_data = ...

    # 定义状态类型
    state_spec = BagStateSpec('my_state', coder=beam.coders.VarIntCoder())

    # 应用ParDo并指定状态
    output = (
        p
        | 'Read Input' >> beam.io.ReadFromText(input_data)
        | 'Apply DoFn' >> beam.ParDo(MyDoFn()).with_stateful_side_inputs(state_spec)
        | 'Write Output' >> beam.io.WriteToText(output_data)
    )