如何使用Apache Beam中的流输入PCollection请求Redis服务器？

import apache_beam as beam
import redis

# 自定义转换函数，将PCollection中的每个元素转换为Redis请求
class RedisRequestTransform(beam.DoFn):
    def __init__(self, redis_host, redis_port):
        self.redis_host = redis_host
        self.redis_port = redis_port

    def start_bundle(self):
        # 建立与Redis服务器的连接
        self.redis_client = redis.Redis(host=self.redis_host, port=self.redis_port)

    def process(self, element):
        # 发送Redis请求
        result = self.redis_client.get(element)
        yield result

# 创建Pipeline对象
p = beam.Pipeline()

# 创建PCollection对象，表示输入的数据流
input_data = p | beam.Create(['key1', 'key2', 'key3'])

# 使用自定义转换函数将PCollection转换为Redis请求
output_data = input_data | beam.ParDo(RedisRequestTransform(redis_host='localhost', redis_port=6379))

# 输出结果
output_data | beam.io.WriteToText('output.txt')

# 运行Pipeline
p.run()