Flink 中使用 Debezium CRUD 的内连接

# 在 Apache Flink 中实现支持 CRUD 操作的表内连接

在基于 Apache Flink 的实时数据集市中，流之间的内连接需要处理来自 Debezium 的所有 CRUD 操作。先前方法忽略了删除和更新操作，导致一对多关系中出现不正确的结果。新实现会解析操作并正确管理状态。

Debezium 消息结构定义了逻辑：op 字段指示操作类型，before 和 after 包含变更前后的数据。

解析 Debezium 消息

Debezium 在 Kafka 中生成具有固定结构的事件：

{
	"op": "(c|r|u|d)",
	"source": { ... },
	"ts_ms": "...",
	"before": [Data, null],
	"after": [Data, null]
}

字段语义：

• c（创建）：before=null，after=Data
• r（读取/快照）：before=null，after=Data
• u（更新）：before=Data，after=Data
• d（删除）：before=Data，after=null

映射器根据 op 提取当前数据。

数据模型映射

Domain 类从 Flink 的 Row 反序列化，同时考虑操作类型：

public class Domain implements Serializable {
  public Integer id;
  public Integer user_id;
  public String domain_name;
  public boolean delete;

  public static Domain fromRow(Row row) {
        Character op = (Character) row.getField(1);

        if (op == null) {
            throw new IllegalStateException("Never should happen, if Debezium feels fine");
        }

        Row domain = (Row) row.getField(op == 'd' ? 0 : 2);

        Integer id = (Integer) domain.getField(0);
        Integer user_id = (Integer) domain.getField(1);
        String domain_name = (String) domain.getField(2);  // ispravleno on String

        return new Domain(id, user_id, domain_name, op == 'd');
    }
}

数据选择逻辑：

• 对于 c/r — 使用 after。
• 对于 d — 始终使用 before。
• 对于 u — 使用 after（详见下一节）。

User 模型同理。

用于内连接的 KeyedCoProcessFunction

InnerJoinFunction 中的核心逻辑使用状态按键（user_id）存储相关记录：

• 当前用户的 ValueState<User>。
• 用户域的 MapState<Integer, Domain>。

当 User 到达时：

• 更新状态。
• 遍历所有域，为每个生成 Output。

当 Domain 到达时：

• 保存到 MapState。
• 如果用户存在，生成 Output。

完整代码：

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.*;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.io.Serializable;

public class InnerJoinFunction extends KeyedCoProcessFunction<Integer, User, Domain, InnerJoinFunction.Output> {
    private MapState<Integer, Domain> domainsState;
    private ValueState<User> usersState;

    @Override
    public void processElement1(final User user, final Context ctx, final Collector<InnerJoinFunction.Output> out) throws Exception {
        usersState.update(user);
      
        for (final Domain domain : domainsState.values()) {
            out.collect(new InnerJoinFunction.Output(
                    user.id,
                    user.firstname,
                    user.lastname,
                    domain.domain_name,
                    user.delete || domain.delete
            ));
        }
    }

    @Override
    public void processElement2(final Domain domain, final Context ctx, final Collector<InnerJoinFunction.Output> out) throws Exception {
        domainsState.put(domain.id, domain);

        final User user = usersState.value();

        if (user != null) {
            out.collect(new InnerJoinFunction.Output(
                    user.id,
                    user.firstname,
                    user.lastname,
                    domain.domain_name,
                    user.delete || domain.delete
            ));
        }
    }

    @Override
    public void open(OpenContext openContext) throws Exception {
        var usersStateDescriptor = new ValueStateDescriptor<>(
                "users",
                User.class
        );
        var domainsStateDescriptor = new MapStateDescriptor<>(
                "domains",
                Integer.class,
                Domain.class
        );
        usersState = getRuntimeContext().getState(usersStateDescriptor);
        domainsState = getRuntimeContext().getMapState(domainsStateDescriptor);

        super.open(openContext);
    }

    public static class Output implements Serializable {
        public Integer user_id;
        public String firstname;
        public String lastname;
        public String domain_name;
        public boolean delete;
        
        // getters and setters omitted
    }
}

实现限制

当前方法正确处理 CRUD，但当更新期间连接键（user_id）发生变化时会失效。状态本地绑定到键，没有重新传输到新键分区。

解决方案见下一部分。

关键点：

• 从 Debezium 解析 op 决定选择 before/after。
• 模型中的 delete 标志确保从数据集市中移除。
• 一对多：遍历 MapState 生成所有组合。
• 对于大数据量，状态后端对性能至关重要。
• 更改连接键需要特殊处理。

— Editorial Team