关于Flink keyBy()函数内部分区及分区数据处理的技术问询

以下代码取自Flink官方文档：

package spendreport;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.source.TransactionSource;

public class FraudDetectionJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Transaction> transactions = env
            .addSource(new TransactionSource())
            .name("transactions");
        
        DataStream<Alert> alerts = transactions
            .keyBy(Transaction::getAccountId)
            .process(new FraudDetector())
            .name("fraud-detector");

        alerts
            .addSink(new AlertSink())
            .name("send-alerts");

        env.execute("Fraud Detection");
    }
}

问题1 #

对transactions数据流调用.keyBy(Transaction::getAccountId)函数，是否会基于account ID对输入数据进行分区，并在每个分区上执行process(new FraudDetector())，如下图所示？

问题2 #

若上述问题答案为是，应如何编写mapper来处理每个分区的流数据？是否需要使用.keyBy(Transaction::getAccountId)？

解答 #

1. 是。keyBy(Transaction::getAccountId)会按照账户ID对数据流做分区，同一账户ID的所有交易数据都会被分配到同一个并行子任务中。后续的process(new FraudDetector())会在每个分区对应的子任务上独立运行，保证同一个账户的所有交易都由同一个FraudDetector实例处理，这样就能维护该账户的状态（如交易频次、金额累计等），实现欺诈检测的核心逻辑，和示意图描述的流程一致。

2. 分两种情况处理：

   • 如果是无状态的映射逻辑（比如仅对单条交易数据做字段转换、格式处理）：不需要使用keyBy，直接在原始数据流上调用map算子即可，这类mapper（MapFunction）不需要感知分区，每条数据独立处理。
   • 如果是需维护账户状态的逻辑（比如统计账户累计交易金额、检测短时间内高频交易）：必须先调用keyBy(Transaction::getAccountId)做分区，然后使用支持状态的算子实现处理逻辑——比如用RichMapFunction（可以在open方法中初始化状态），或者更灵活的ProcessFunction（示例中的FraudDetector就是这类）。只有经过keyBy，每个账户的状态才会被隔离在对应分区中，避免不同账户的数据互相干扰。

内容的提问来源于stack exchange，提问作者overexchange