// 旧方式 - 使用已不推荐使用的 DataSet API (Java 8)
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<String> data = env.fromElements("a", "b", "c");
data.print();

// 新方式 - 使用 DataStream API 处理批数据 (Java 8)
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 关键设置：明确指定为 BATCH 模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

// 读取有界数据源（例如文件）作为流
DataStream<String> data = env.readTextFile("path/to/your/batch/data.txt");

data.print();
env.execute("Batch Job with DataStream API");

// 旧方式 - 使用不推荐的自定义 SinkFunction (Java 8)
data.addSink(new SinkFunction<String>() {
    @Override
    public void invoke(String value, Context context) throws Exception {
        // 简单直接地输出，难以保证端到端的恰好一次语义
        System.out.println(value);
    }
});

// 新方式 - 使用新的 SinkTo API (Java 8)
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.avro.AvroParquetWriters;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

// 1. 定义一个新的 Sink，例如 FileSink
FileSink<String> fileSink = FileSink
    .forRowFormat(new Path("output/path"), new SimpleStringEncoder<String>("UTF-8"))
    .withRollingPolicy(
        DefaultRollingPolicy.builder()
            .withRolloverInterval(Duration.ofMinutes(15).toMillis())
            .build()
    )
    .build();

// 2. 使用新的 sinkTo 方法
data.sinkTo(fileSink);


// 另一个例子：使用 KafkaSink
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;

KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka-broker:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-topic")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build()
    )
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // 轻松设置交付保证
    .setTransactionalIdPrefix("my-app-")
    .build();

data.sinkTo(kafkaSink);

// 旧方式 - 已弃用
DataStream<String> stream = env.addSource(new SourceFunction<String>() {
    private volatile boolean isRunning = true;
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRunning) {
            // ... 产生数据
        }
    }
    @Override
    public void cancel() {
        isRunning = false;
    }
});

// 新方式 - 使用新的 Source 实现 (例如 Kafka)
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka-broker:9092")
    .setTopics("input-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> data = env.fromSource(
    source,
    WatermarkStrategy.noWatermarks(), // 指定水印策略
    "Kafka Source"
);

-- 旧 Group Window 语法 (已过时)
SELECT
  TUMBLE_START(ts, INTERVAL '1' HOUR),
  user,
  COUNT(url)
FROM clicks
GROUP BY TUMBLE(ts, INTERVAL '1' HOUR), user

-- 新 Window TVF 语法 (推荐)
SELECT
  window_start,
  window_end,
  user,
  COUNT(url)
FROM TABLE(
  TUMBLE(TABLE clicks, DESCRIPTOR(ts), INTERVAL '1' HOUR)
)
GROUP BY window_start, window_end, user