如何在Apache beam中对时间窗口内的元素进行计数，并在计数达到某个阈值时发出数据？

import apache_beam as beam
from apache_beam.transforms.trigger import AfterCount

class CountElements(beam.DoFn):
    def __init__(self, threshold):
        self.threshold = threshold
        self.count = 0

    def process(self, element, window=beam.DoFn.WindowParam):
        self.count += 1
        if self.count >= self.threshold:
            yield element
            self.count = 0

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

# 从某个数据源读取数据，例如从Kafka读取
data = p | beam.io.ReadFromKafka(...)

# 将数据按照时间窗口进行分组
windowed_data = data | beam.WindowInto(beam.window.FixedWindows(10))

# 对时间窗口内的元素进行计数，并在计数达到阈值时发出数据
counted_data = windowed_data | beam.ParDo(CountElements(5))

# 将结果写入某个目的地，例如写入Kafka
counted_data | beam.io.WriteToKafka(...)

# 运行Pipeline
p.run()