tongchenkeji
发表于：0次点击

flink中如何能够产生-D的change log的呢？[阿里云]

flink中如何能够产生-D的change log的呢？

实际的业务场景是，我这边有一个滑动窗口，但是我希望根据第二次滑动后窗口产生的数据和第一次滑动后窗口产生的数据做对比，按照主键做对比，新增数据做insert，消失的数据做delete，分别产生需要insert和delete的数据。

但是网上不管怎么搜，都只有产生+I -U +U类型的example，就是没有产生-D的example

「点点赞赏，手留余香」

    还没有人赞赏，快来当第一个赞赏的人吧！
阿里云# 实时计算 Flink版3179# 流计算2236
=====这是一个广告位，招租中，联系qq 78315851====
阿里云
Flink日期减去24小时 怎么实现？需要保留时分秒[阿里云]
阿里云
flink 可以让他自动重启的时候都无状态重启吗？[阿里云]
1 条回复 A 作者 M 管理员

  1. 在Flink中，要产生-D的change log，可以通过自定义一个ChangelogProducer来实现。以下是一个简单的示例：

    首先，创建一个自定义的ChangelogProducer类，继承自org.apache.flink.streaming.api.functions.source.SourceFunction>：

    import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.streaming.api.watermark.Watermark;import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import org.apache.flink.util.Collector;public class CustomChangelogProducer<T> implements SourceFunction<ChangeEvent<T>> {    private final Supplier<T> dataSupplier;    private final long windowSize;    private T currentData;    public CustomChangelogProducer(Supplier<T> dataSupplier, long windowSize) {        this.dataSupplier = dataSupplier;        this.windowSize = windowSize;    }    @Override    public void run(SourceContext<ChangeEvent<T>> ctx) throws Exception {        while (true) {            long currentTime = System.currentTimeMillis();            T newData = dataSupplier.get();            if (newData != null) {                currentData = newData;                ctx.collect(new ChangeEvent<>(currentTime, currentTime + windowSize, currentData));            }            Thread.sleep(1000);        }    }    @Override    public void cancel() {        // No-op    }    @Override    public boolean isBounded() {        return false;    }    @Override    public void onElement(ChangeEvent<T> element, Context context, Collector<ChangeEvent<T>> collector) throws Exception {        // Do nothing    }    @Override    public void onTimeout(long timeout, TimeWindow window, Context context) throws Exception {        // Do nothing    }    @Override    public void onWatermark(Watermark mark) throws Exception {        // Do nothing    }}

    然后，在你的Flink程序中使用这个自定义的ChangelogProducer：

    import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.source.SourceFunction;import org.apache.flink.streaming.api.watermark.WatermarkStrategy;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.streaming.api.windowing.windows.TimeWindow;import org.apache.flink.util.Collector;public class Main {    public static void main(String[] args) throws Exception {        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // 使用自定义的ChangelogProducer        DataStream<String> inputStream = env.addSource(new CustomChangelogProducer<>(() -> "example", 1000L));        // 对输入数据进行处理，例如按照主键进行对比，生成insert和delete的数据        // ...        env.execute("Flink Changelog Example");    }}

    这样，你就可以在Flink程序中产生-D的change log了。注意，这个示例仅用于演示目的，实际应用中可能需要根据具体需求进行调整。

    已喜欢 已反对

  2. 在Apache Flink中，你可以使用Table API或SQL来处理数据流，并生成Change Log（变更日志）。这些变更日志可以包含INSERT、UPDATE和DELETE操作。通常情况下，Flink的CDC（Change Data Capture）功能是针对数据库源的数据变更捕获。

    要实现你所描述的场景，你需要将滑动窗口产生的结果转换为I/U/D形式的变更日志。以下是大致的步骤：

    1. 创建表结构：
      首先，定义一个与你要比较的数据相对应的表结构。这个表结构应该包含主键列和其他属性列。

    2. 从数据源读取数据：
      使用Table API或SQL从你的数据源读取数据，并将其注册为一个临时表。

    3. 应用滑动窗口：
      使用GROUP BY和TUMBLE等SQL语句对数据进行滑动窗口分组，并计算每个窗口内的数据。

    4. 计算差异：
      通过对比两次滑动窗口的结果，找出新增、更新和删除的数据。这可能需要编写一些自定义的UDF（用户自定义函数），以便根据主键值进行比较。

    5. 产生变更日志：
      将计算出的差异转换为INSERT、UPDATE和DELETE操作。对于INSERT和UPDATE，你可以直接使用INSERT INTO和UPDATE SQL语句。对于DELETE操作，你需要找到上次窗口中存在的但当前窗口中不存在的数据，并生成相应的DELETE语句。

    6. 输出变更日志：
      最后，将产生的变更日志写入到另一个表或者输出到外部系统。

    由于Flink没有直接提供这种场景的内置操作，你可能需要编写一些自定义代码来实现上述逻辑。这里是一个简化的示例，展示了如何在Flink Table API中使用row_number()窗口函数和聚合操作来找到前一次窗口与当前窗口之间的差异：

    // 假设我们有一个名为input的原始表，其中包含两个字段：id（主键）和value// 定义一个临时视图，用于存储前一次窗口的结果tableEnv.createTemporaryView("previous_window", previousWindowResult);// 对输入表应用滑动窗口，并与上一次窗口的结果进行JOINTable result = input    .window(Tumble.over(lit(5).minutes()).on($("timestamp")).as("w"))    .groupBy($("w"), $("id"))    .select(        $("id"),        $("value").as("current_value"),        $("previous_window.value").as("previous_value")    );// 对比当前窗口和上一次窗口的结果，找出新增、更新和删除的数据result    .where($("previous_value").isNull())    .insertInto("inserts");result    .where($("previous_value").isNotNull())    .where($("current_value").notEqual($("previous_value")))    .insertInto("updates");result    .where($("current_value").isNull())    .insertInto("deletes");
    已喜欢 已反对