Apache Flink 使用教程：从入门到精通

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 1. 读取数据源
DataStream<String> source = env.fromSource(...);

// 2. 数据转换
DataStream<String> result = source
    .map(...)
    .keyBy(...)
    .process(...);

// 3. 写入数据汇
result.sinkTo(...);

// 4. 提交执行
env.execute("MyFlinkJob");

java -version
# openjdk version "17.0.x"

# 下载并解压
tar -xzf flink-2.2.0-bin-scala_2.12.tgz
cd flink-2.2.0

./bin/start-cluster.sh
# 启动后访问 http://localhost:8081 查看 Web UI

./bin/stop-cluster.sh

mvn archetype:generate \
  -DarchetypeGroupId=org.apache.flink \
  -DarchetypeArtifactId=flink-quickstart-java \
  -DarchetypeVersion=2.2.0

curl https://flink.apache.org/q/quickstart.sh | bash -s 2.2.0

<properties>
    <flink.version>2.2.0</flink.version>
    <scala.binary.version>_2.12</scala.binary.version>
    <java.version>11</java.version>
</properties>

<dependencies>
    <!-- DataStream API -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <!-- 客户端（本地执行与提交） -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <!-- Kafka 连接器示例（需打进 fat jar） -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordCount {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        final StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // 从 socket 读取文本流
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // 拆词、分组、计数
        DataStream<Tuple2<String, Integer>> counts = text
            // 过滤空行
            .filter(line -> !line.isEmpty())
            // 按空格拆分，每个词映射为 (word, 1)
            .flatMap((line, out) -> {
                for (String word : line.toLowerCase().split("\\s+")) {
                    out.collect(Tuple2.of(word, 1));
                }
            })
            .returns(Tuple2.class)
            // 按单词分组（keyBy 的第一个字段）
            .keyBy(value -> value.f0)
            // 滚动求和
            .sum(1);

        counts.print();

        env.execute("Socket WordCount");
    }
}

# 终端 1：启动 socket 服务
nc -lk 9999

# 终端 2：运行作业（IDE 直接运行 main 方法即可）
# 然后在终端 1 输入文字，观察终端 2 输出

# 打包
mvn clean package

# 提交到本地集群
./bin/flink run target/your-job-0.1.jar

# 或通过 Web UI 上传 jar 提交

./bin/flink list              # 列出作业
./bin/flink cancel <jobId>    # 取消作业

// 从集合创建（测试用）
DataStream<Event> events = env.fromCollection(eventList);

// 从文件创建（有界流，批模式）
DataStream<String> lines = env.readTextFile("file:///path/to/file");

// 从 Kafka 创建（最常用的实时源）
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setTopics("input-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> stream = env.fromSource(
    kafkaSource,
    WatermarkStrategy.noWatermarks(),
    "kafka-source"
);

public class CountAverage extends KeyedProcessFunction<String, Tuple2<String, Long>, Double> {

    // 值状态：保存单个值
    private ValueState<Long> countState;
    // 列表状态：保存一组值
    private ListState<Long> valuesState;
    // 映射状态：键值对
    private MapState<String, Long> mapState;
    // 归约状态：自动归约
    private ReducingState<Long> sumState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> countDesc =
            new ValueStateDescriptor<>("count", Long.class);
        countState = getRuntimeContext().getState(countDesc);

        ListStateDescriptor<Long> valuesDesc =
            new ListStateDescriptor<>("values", Long.class);
        valuesState = getRuntimeContext().getListState(valuesDesc);
    }

    @Override
    public void processElement(
            Tuple2<String, Long> value,
            Context ctx,
            Collector<Double> out) throws Exception {

        long count = countState.value() == null ? 0 : countState.value();
        count++;
        countState.update(count);
        valuesState.add(value.f1);

        // 计算平均值
        long sum = 0;
        for (Long v : valuesState.get()) {
            sum += v;
        }
        out.collect((double) sum / count);
    }
}

public class TimeoutAlert extends KeyedProcessFunction<String, Order, Alert> {

    private ValueState<Long> timerState;

    @Override
    public void open(Configuration cfg) {
        timerState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("timer", Long.class));
    }

    @Override
    public void processElement(Order order, Context ctx, Collector<Alert> out)
            throws Exception {
        // 注册一个 5 分钟后的定时器（基于事件时间）
        long timeout = ctx.timestamp() + 5 * 60 * 1000L;
        ctx.timerService().registerEventTimeTimer(timeout);
        timerState.update(timeout);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out)
            throws Exception {
        // 定时器触发，说明 5 分钟内没有后续事件，发出告警
        out.collect(new Alert(ctx.getCurrentKey(), "timeout"));
    }
}

// 侧流输出：把不符合条件的数据分流到旁路
OutputTag<Order> lateTag = new OutputTag<Order>("late-orders"){};

SingleOutputStreamOperator<Order> mainStream = orders.process(
    new ProcessFunction<Order, Order>() {
        @Override
        public void processElement(Order order, Context ctx, Collector<Order> out) {
            if (order.isLate()) {
                ctx.output(lateTag, order);  // 发到侧流
            } else {
                out.collect(order);          // 留在主流
            }
        }
    });

// 获取侧流
DataStream<Order> lateStream = mainStream.getSideOutput(lateTag);

import org.apache.flink.table.api.*;

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()   // 或 inBatchMode()
    .build();

TableEnvironment tEnv = TableEnvironment.create(settings);

-- 定义 Kafka 源表
CREATE TABLE orders (
    order_id    STRING,
    user_id     BIGINT,
    amount      DECIMAL(10, 2),
    order_time  TIMESTAMP(3),
    -- 用 WATERMARK 定义事件时间和水位线策略
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'order-group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);

-- 定义 MySQL 结果表
CREATE TABLE order_stats (
    user_id   BIGINT,
    total_amt DECIMAL(10, 2),
    cnt       BIGINT,
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://localhost:3306/report',
    'table-name' = 'order_stats',
    'username' = 'root',
    'password' = 'pwd'
);

-- 实时统计每个用户的订单总额和订单数
INSERT INTO order_stats
SELECT
    user_id,
    SUM(amount) AS total_amt,
    COUNT(*) AS cnt
FROM orders
GROUP BY user_id;

// Table API 查询
Table resultTable = tEnv.sqlQuery(
    "SELECT user_id, SUM(amount) FROM orders GROUP BY user_id");

// Table 转 DataStream（结果流是 changelog）
DataStream<Row> changelogStream = tEnv
    .toChangelogStream(resultTable);

// DataStream 转 Table
tEnv.createTemporaryView("my_stream", dataStream);

DataStream<Event> withTimestamps = env
    .fromSource(kafkaSource,
        WatermarkStrategy
            .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, ts) -> event.getTimestamp()),
        "kafka-source");

-- 滚动窗口：每 1 分钟统计一次订单量
SELECT
    window_start,
    window_end,
    COUNT(*) AS order_cnt,
    SUM(amount) AS total
FROM TABLE(
    TUMBLE(TABLE orders, DESCRIPTOR(order_time), INTERVAL '1' MINUTE)
)
GROUP BY window_start, window_end;

-- 滑动窗口：每 30 秒统计过去 1 分钟的数据
SELECT
    window_start, window_end, user_id, SUM(amount)
FROM TABLE(
    HOP(TABLE orders, DESCRIPTOR(order_time),
        INTERVAL '30' SECOND, INTERVAL '1' MINUTE)
)
GROUP BY window_start, window_end, user_id;

-- 会话窗口：30 分钟无活动则切分会话
SELECT
    window_start, window_end, user_id, COUNT(*)
FROM TABLE(
    SESSION(TABLE orders, DESCRIPTOR(order_time), INTERVAL '30' MINUTE)
)
GROUP BY window_start, window_end, user_id;

DataStream<...> result = input
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.seconds(10))   // 允许 10 秒迟到
    .sideOutputLateData(lateTag)         // 更迟的送到侧流
    .sum("amount");

# 状态后端类型
state.backend.type: rocksdb
# Checkpoint 存储目录
execution.checkpointing.dir: hdfs://namenode:40010/flink/checkpoints
# 开启增量 Checkpoint（仅 RocksDB/ForSt 支持）
execution.checkpointing.incremental: true

Configuration config = new Configuration();
config.set(StateBackendOptions.STATE_BACKEND, "rocksdb");
env.configure(config);

env.enableCheckpointing(60_000);  // 每 60 秒一次

// 更精细的配置
CheckpointConfig config = env.getCheckpointConfig();
config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
config.setMinPauseBetweenCheckpoints(30_000);  // 两次 checkpoint 间隔
config.setCheckpointTimeout(600_000);          // 超时时间
config.setMaxConcurrentCheckpoints(1);         // 最大并发数
config.setExternalizedCheckpointCleanup(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

# 开启 unaligned checkpoint（反压场景）
execution.checkpointing.unaligned.enabled: true
# 开启 buffer debloating
execution.buffer-debloat.enabled: true

# 触发 savepoint
./bin/flink savepoint <jobId> hdfs:///savepoints/

# 从 savepoint 恢复（取消并重启）
./bin/flink run -s hdfs:///savepoints/savepoint-xxxx -c MyApp job.jar

# 从 savepoint 恢复并允许丢弃状态（有状态算子删除时）
./bin/flink run -s hdfs:///savepoints/savepoint-xxxx --allowNonRestoredState ...

-- Source：读取 Kafka
CREATE TABLE kafka_source (
    `user_id`   BIGINT,
    `item_id`   BIGINT,
    `behavior`  STRING,
    `ts`        TIMESTAMP(3),
    WATERMARK FOR ts AS ts - INTERVAL '3' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_behavior',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'behavior-group',
    'scan.startup.mode' = 'group-offsets',
    'format' = 'json'
);

-- Sink：写入 Kafka
CREATE TABLE kafka_sink (
    `window_end`  TIMESTAMP(3),
    `behavior`    STRING,
    `cnt`         BIGINT
) WITH (
    'connector' = 'kafka',
    'topic' = 'behavior_stats',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json'
);

-- 管道：实时统计每分钟各行为的次数
INSERT INTO kafka_sink
SELECT
    window_end, behavior, COUNT(*) AS cnt
FROM TABLE(
    TUMBLE(TABLE kafka_source, DESCRIPTOR(ts), INTERVAL '1' MINUTE)
)
GROUP BY window_end, behavior;

CREATE TABLE products (
    id          INT,
    name        STRING,
    description STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'mysql-host',
    'port' = '3306',
    'username' = 'flinkuser',
    'password' = 'pwd',
    'database-name' = 'inventory',
    'table-name' = 'products'
);

-- 创建 Paimon 表
CREATE TABLE paimon_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10, 2),
    dt STRING,
    PRIMARY KEY (order_id, dt) NOT ENFORCED
) WITH (
    'connector' = 'paimon',
    'warehouse' = 's3://my-bucket/warehouse',
    'path' = 'orders'
);

-- 实时写入
INSERT INTO paimon_orders
SELECT order_id, user_id, amount, DATE_FORMAT(order_time, 'yyyy-MM-dd')
FROM kafka_source;

CREATE MATERIALIZED TABLE daily_sales
    PARTITIONED BY (dt)
    FRESHNESS = INTERVAL '1' HOUR
    AS SELECT
        dt,
        product_id,
        SUM(amount) AS total
    FROM orders
    GROUP BY dt, product_id;

-- 注册 OpenAI 模型
CREATE MODEL `translator`
INPUT (text STRING)
OUTPUT (response STRING)
WITH (
    'provider' = 'openai',
    'endpoint' = 'https://api.openai.com/v1/chat/completions',
    'api-key' = 'sk-xxx',
    'model' = 'gpt-4o',
    'system-prompt' = 'translate to Chinese'
);

-- 在流式查询中实时调用模型
SELECT * FROM ML_PREDICT(
    TABLE input_logs,
    MODEL translator,
    DESCRIPTOR(text),
    MAP['async', 'true', 'timeout', '100s']
);

SELECT * FROM
    input_table,
    LATERAL VECTOR_SEARCH(
        TABLE vector_table,
        input_table.query_vector,
        DESCRIPTOR(index_column),
        10,
        MAP['async', 'true', 'timeout', '100s']
    );

// 创建模型
tEnv.createModel("my_model",
    ModelDescriptor.forProvider("openai")
        .inputSchema(Schema.newBuilder().column("input", DataTypes.STRING()).build())
        .outputSchema(Schema.newBuilder().column("output", DataTypes.STRING()).build())
        .option("model", "gpt-4.1")
        .option("api-key", "sk-xxx")
        .build());

// 执行推理（支持异步）
Model model = tEnv.fromModel("my_model");
Table result = model.predict(myTable, ColumnList.of("text"),
    Map.of("async", "true"));

CREATE TABLE raw_events (
    id INT,
    payload VARIANT  -- 直接存储动态 JSON
) WITH ('connector' = 'paimon', ...);

-- 用 PARSE_JSON 把字符串转为 VARIANT
INSERT INTO raw_events
SELECT id, PARSE_JSON(json_str) FROM kafka_source;

public static class GreetingWithMemory extends ProcessTableFunction<String> {
    public static class CountState { public long counter = 0L; }

    public void eval(@StateHint CountState state,
                     @ArgumentHint(SET_SEMANTIC_TABLE) Row input) {
        state.counter++;
        collect("Hello " + input.getFieldAs("name")
            + ", your " + state.counter + " time?");
    }
}

SELECT * FROM GreetingWithMemory(TABLE Names PARTITION BY name)

SET 'table.optimizer.multi-join.enabled' = 'true';

helm repo add flink-operator \
    https://downloads.apache.org/flink/flink-kubernetes-operator-1.14.0/
helm install flink-operator flink-operator/flink-kubernetes-operator

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: my-flink-job
spec:
  image: flink:2.2.0
  flinkVersion: v2_2
  flinkConfiguration:
    state.backend.type: rocksdb
    execution.checkpointing.interval: 60s
    execution.checkpointing.dir: s3://my-bucket/checkpoints
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "4096m"
      cpu: 2
  job:
    jarURI: local:///opt/flink/usrlib/my-job.jar
    parallelism: 4
    entryClass: com.example.MyFlinkJob
    state: running

high-availability.type: zookeeper
high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181
high-availability.zookeeper.path.root: /flink
high-availability.storageDir: hdfs:///flink/ha/

metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
metrics.reporter.prom.port: 9999

state.backend.rocksdb.memory.managed: true
state.backend.rocksdb.memory.write-buffer-ratio: 0.5
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1

-- 自动开启 local-global 两阶段聚合
SET 'table.optimizer.agg-phase-strategy' = 'AUTO';

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '5s';
SET 'table.exec.mini-batch.size' = '5000';

-- ODS：原始日志从 Kafka 读入
CREATE TABLE ods_logs (
    log_id STRING,
    user_id BIGINT,
    event STRING,
    properties STRING,
    ts TIMESTAMP(3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka', ...);

-- DWS：每分钟按事件类型聚合
CREATE TABLE dws_event_min (
    window_end TIMESTAMP(3),
    event STRING,
    user_cnt BIGINT,
    event_cnt BIGINT
) WITH ('connector' = 'paimon', 'warehouse' = 's3://wh/dws', ...);

INSERT INTO dws_event_min
SELECT
    window_end,
    event,
    COUNT(DISTINCT user_id) AS user_cnt,
    COUNT(*) AS event_cnt
FROM TABLE(
    TUMBLE(TABLE ods_logs, DESCRIPTOR(ts), INTERVAL '1' MINUTE)
)
GROUP BY window_end, event;

-- 1. 实时计算用户近 5 分钟交易频率
CREATE TABLE user_txn_freq (
    user_id BIGINT,
    window_end TIMESTAMP(3),
    txn_cnt BIGINT,
    total_amount DECIMAL(10,2)
) WITH ('connector' = 'kafka', ...);

INSERT INTO user_txn_freq
SELECT user_id, window_end, COUNT(*), SUM(amount)
FROM TABLE(
    TUMBLE(TABLE transactions, DESCRIPTOR(txn_time), INTERVAL '5' MINUTE)
)
GROUP BY user_id, window_end;

-- 2. 用大模型对高频交易做风险研判
CREATE MODEL risk_model
INPUT (desc STRING)
OUTPUT (risk_level STRING)
WITH (
    'provider' = 'openai',
    'model' = 'gpt-4o',
    'system-prompt' = '判断该交易行为的风险等级：低/中/高',
    'api-key' = 'sk-xxx'
);

-- 3. 把可疑交易描述喂给模型，实时输出风险等级
SELECT * FROM ML_PREDICT(
    TABLE (
        SELECT CONCAT('用户', CAST(user_id AS STRING),
               '5分钟内交易', CAST(txn_cnt AS STRING),
               '笔，金额', CAST(total_amount AS STRING)) AS desc
        FROM user_txn_freq WHERE txn_cnt > 10
    ),
    MODEL risk_model,
    DESCRIPTOR(desc),
    MAP['async', 'true', 'timeout', '30s']
);