如何使用Apache Flink按datetime分区在HDFS上写入parquet文件？

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink;
import org.apache.flink.streaming.connectors.fs.bucketing.DateTimeBucketer;

public class FlinkParquetWriter {
    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从数据源读取数据流
        DataStream<Tuple2<String, Integer>> dataStream = env.fromElements(
                new Tuple2<>("A", 1),
                new Tuple2<>("B", 2),
                new Tuple2<>("C", 3)
        );

        // 按照每小时的时间窗口进行分区
        DataStream<Tuple2<String, Integer>> windowedStream = dataStream
                .keyBy(0)
                .timeWindow(Time.hours(1))
                .sum(1);

        // 将数据写入HDFS上的parquet文件
        BucketingSink<Tuple2<String, Integer>> sink = new BucketingSink<>("hdfs://localhost:9000/path/to/parquet/files");
        sink.setBucketer(new DateTimeBucketer<>("yyyy-MM-dd/HH"));
        sink.setWriter(new ParquetWriter<>());
        windowedStream.addSink(sink);

        // 执行任务
        env.execute("Flink Parquet Writer");
    }
}