我正在尝试将消息从kafka消费者流式传输到谷歌云存储，使用apache光束有30秒的窗口。用于beam_nuggets.io从kafka主题中读取。然而，我无法为每个窗口编写独特的镶木地板文件GCS。你可以在下面看到我的代码：'

import apache_beam as beam
from apache_beam.transforms.trigger import AfterAny, AfterCount, AfterProcessingTime, AfterWatermark, Repeatedly

from apache_beam.portability.api.beam_runner_api_pb2 import AccumulationMode
from apache_beam.options.pipeline_options import PipelineOptions
from beam_nuggets.io import kafkaio
import json
from datetime import datetime
import pandas as pd
import config as conf
import apache_beam.transforms.window as window
consumer_config = {"topic": "Uswrite",
                   "bootstrap_servers": "*.*.*.*:9092",
                   "group_id": "notification_consumer_group_33"}
folder_name = datetime.now().strftime('%Y-%m-%d')
def format_result(consume_message):
    data = json.loads(consume_message[1])
    file_name = datetime.now().strftime("%Y_%m_%d-%I_%M_%S")
    df = pd.DataFrame(data).T #, orient='index'
    df.to_parquet(f'gs://{conf.gcs}/{folder_name}/{file_name}.parquet',
               storage_options={"token": "gcp.json"}, engine='fastparquet')
    print(consume_message)
with beam.Pipeline(options=PipelineOptions()) as p:
    consumer_message = (p | "Reading messages from Kafka" >> kafkaio.KafkaConsume(consumer_config=consumer_config)
                          | 'Windowing' >> beam.WindowInto(window.FixedWindows(30),
                                            trigger=AfterProcessingTime(30),
                                            allowed_lateness=900,
                                           accumulation_mode=AccumulationMode.ACCUMULATING)
                          | 'CombineGlobally' >> beam.Map(format_result))

# window.FixedWindows(30),trigger=beam.transforms.trigger.AfterProcessingTime(30),
# accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING
# allowed_lateness=100,CombineGlobally(format_result).without_defaults() allowed_lateness=30,

使用上面的代码，为每条消息生成一个新的拼花文件。我想做的是按30秒的窗口对消息进行分组，并为每个窗口生成一个拼花文件。我尝试了下面的不同配置，但没有成功：Beam. CombineGlobally（format_result）.without_defaults（））而不是beam.Map（format_result））Beam.ParDo（format_result））此外，我还有几个问题：

1. 即使我将偏移量设置为auto. offset.set：最早，kafka生产者也会从最后一条消息开始读取，即使我更改了消费者组并且找不到原因。
2. 另外，我对触发器、allowed_lateness、accumulation_mode的用法感到困惑。我不确定我是否需要它们来完成这项任务。

正如您在上面的代码块中看到的，我也尝试使用这些参数，但没有帮助。

我到处搜索，但找不到一个解释此用例的示例。