实时计算 Flink版产品使用合集之什么情况作业会被认为是有限流作业呢二者该怎么区分

// 定义输入流表和历史表的DDL
String inputTableDDL = "CREATE TABLE input_table (id INT, name STRING, updateTime TIMESTAMP(3)) " +
        "WITH (...)";
String historyTableDDL = "CREATE TABLE history_table (id INT, name STRING, updateTime TIMESTAMP(3)) " +
        "WITH (...)";
// 将DDL注册为表
tableEnv.executeSql(inputTableDDL);
tableEnv.executeSql(historyTableDDL);
// 执行时态表连接查询
String query = "SELECT * FROM input_table FOR SYSTEM_TIME AS OF history_table.updateTime AS i, history_table " +
        "WHERE i.id = history_table.id";
Table result = tableEnv.sqlQuery(query);
// 处理查询结果
DataStream<Row> output = tableEnv.toAppendStream(result, Row.class);

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.streaming.api.datastream.DataStream;
public class MyCEPExample {
    public static void main(String[] args) throws Exception {
        // 假设我们有一个 DataStream 数据源
        DataStream<String> input = ...;
        // 定义一个 CEP 模式
        Pattern<String, String> pattern = Pattern.<String>begin("start").where(new SimpleCondition<String>() {
            @Override
            public boolean filter(String value) {
                return value.startsWith("start");
            }
        }).next("end").where(new SimpleCondition<String>() {
            @Override
            public boolean filter(String value) {
                return value.endsWith("end");
            }
        });
        // 创建一个 PatternStream 对象
        PatternStream<String> patternStream = CEP.pattern(input, pattern);
        // 使用 select 方法处理每个匹配的结果
        DataStream<String> result = patternStream.select(new PatternSelectFunction<String, String>() {
            @Override
            public String select(Map<String, List<String>> pattern) {
                // 这里可以访问到所有匹配上的事件
                List<String> startEvents = pattern.get("start");
                List<String> endEvents = pattern.get("end");
                // 输出当前匹配的数据
                System.out.println("Matched events: " + startEvents + ", " + endEvents);
                // 返回一个代表匹配结果的字符串
                return "matched";
            }
        });
        // 提交并运行作业
        env.execute("My CEP Example");
    }
}