Flink 流式处理基础

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
public class OrderEvent {

    private String eventId;
    private String orderId;
    private Long shopId;
    private String eventType;
    private java.math.BigDecimal amount;
    private long eventTime;

    public String getEventId() {
        return eventId;
    }

    public void setEventId(String eventId) {
        this.eventId = eventId;
    }

    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public Long getShopId() {
        return shopId;
    }

    public void setShopId(Long shopId) {
        this.shopId = shopId;
    }

    public String getEventType() {
        return eventType;
    }

    public void setEventType(String eventType) {
        this.eventType = eventType;
    }

    public java.math.BigDecimal getAmount() {
        return amount;
    }

    public void setAmount(java.math.BigDecimal amount) {
        this.amount = amount;
    }

    public long getEventTime() {
        return eventTime;
    }

    public void setEventTime(long eventTime) {
        this.eventTime = eventTime;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@Service
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishPaidEvent(String orderId, Long shopId, BigDecimal amount, Instant paidAt) {
        OrderEvent event = new OrderEvent();
        event.setEventId(UUID.randomUUID().toString());
        event.setOrderId(orderId);
        event.setShopId(shopId);
        event.setEventType("PAID");
        event.setAmount(amount);
        event.setEventTime(paidAt.toEpochMilli());

        kafkaTemplate.send("order-events", String.valueOf(shopId), event);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
public class ShopMinuteGmvJob {

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

        env.enableCheckpointing(30000L);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000L);
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("localhost:9092")
                .setTopics("order-events")
                .setGroupId("shop-minute-gmv")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        WatermarkStrategy<OrderEvent> watermarkStrategy =
                WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> event.getEventTime());

        DataStream<OrderEvent> orderEventStream = env
                .fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-order-source")
                .map(Jsons::readOrderEvent)
                .assignTimestampsAndWatermarks(watermarkStrategy)
                .filter(event -> "PAID".equals(event.getEventType()));

        DataStream<OrderEvent> deduplicatedStream = orderEventStream
                .keyBy(OrderEvent::getEventId)
                .process(new EventIdDeduplicateFunction());

        DataStream<ShopMinuteGmv> resultStream = deduplicatedStream
                .keyBy(OrderEvent::getShopId)
                .window(TumblingEventTimeWindows.of(Time.minutes(1)))
                .allowedLateness(Time.seconds(30))
                .aggregate(new ShopMinuteGmvAggregate(), new ShopMinuteGmvWindowFunction());

        resultStream.addSink(MySqlUpsertSink.build());

        env.execute("shop-minute-gmv-job");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public class EventIdDeduplicateFunction
        extends KeyedProcessFunction<String, OrderEvent, OrderEvent> {

    private transient ValueState<Boolean> seenState;

    @Override
    public void open(Configuration parameters) {
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(org.apache.flink.api.common.time.Time.hours(2))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .neverReturnExpired()
                .build();

        ValueStateDescriptor<Boolean> descriptor =
                new ValueStateDescriptor<>("seen-event", Types.BOOLEAN);
        descriptor.enableTimeToLive(ttlConfig);
        seenState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(
            OrderEvent value,
            Context ctx,
            Collector<OrderEvent> out) throws Exception {

        if (Boolean.TRUE.equals(seenState.value())) {
            return;
        }

        seenState.update(Boolean.TRUE);
        out.collect(value);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class ShopMinuteGmv {

    private Long shopId;
    private Instant windowStart;
    private Instant windowEnd;
    private long paidOrderCount;
    private BigDecimal paidAmount;

    public void addAmount(BigDecimal delta) {
        if (paidAmount == null) {
            paidAmount = BigDecimal.ZERO;
        }
        paidAmount = paidAmount.add(delta);
    }

    // getter / setter 省略
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public class ShopMinuteGmvAggregate
        implements AggregateFunction<OrderEvent, ShopMinuteGmv, ShopMinuteGmv> {

    @Override
    public ShopMinuteGmv createAccumulator() {
        ShopMinuteGmv acc = new ShopMinuteGmv();
        acc.setPaidAmount(BigDecimal.ZERO);
        return acc;
    }

    @Override
    public ShopMinuteGmv add(OrderEvent value, ShopMinuteGmv acc) {
        acc.setShopId(value.getShopId());
        acc.setPaidOrderCount(acc.getPaidOrderCount() + 1);
        acc.addAmount(value.getAmount());
        return acc;
    }

    @Override
    public ShopMinuteGmv getResult(ShopMinuteGmv acc) {
        return acc;
    }

    @Override
    public ShopMinuteGmv merge(ShopMinuteGmv a, ShopMinuteGmv b) {
        a.setPaidOrderCount(a.getPaidOrderCount() + b.getPaidOrderCount());
        a.addAmount(b.getPaidAmount());
        return a;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public class ShopMinuteGmvWindowFunction
        extends ProcessWindowFunction<ShopMinuteGmv, ShopMinuteGmv, Long, TimeWindow> {

    @Override
    public void process(
            Long shopId,
            Context context,
            Iterable<ShopMinuteGmv> elements,
            Collector<ShopMinuteGmv> out) {

        ShopMinuteGmv result = elements.iterator().next();
        result.setShopId(shopId);
        result.setWindowStart(Instant.ofEpochMilli(context.window().getStart()));
        result.setWindowEnd(Instant.ofEpochMilli(context.window().getEnd()));
        out.collect(result);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public final class MySqlUpsertSink {

    private MySqlUpsertSink() {
    }

    public static SinkFunction<ShopMinuteGmv> build() {
        return JdbcSink.sink(
                """
                INSERT INTO ads_shop_gmv_1m
                (window_start, window_end, shop_id, paid_order_count, paid_amount, updated_at)
                VALUES (?, ?, ?, ?, ?, NOW())
                ON DUPLICATE KEY UPDATE
                    paid_order_count = VALUES(paid_order_count),
                    paid_amount = VALUES(paid_amount),
                    updated_at = NOW()
                """,
                (ps, value) -> {
                    ps.setTimestamp(1, Timestamp.from(value.getWindowStart()));
                    ps.setTimestamp(2, Timestamp.from(value.getWindowEnd()));
                    ps.setLong(3, value.getShopId());
                    ps.setLong(4, value.getPaidOrderCount());
                    ps.setBigDecimal(5, value.getPaidAmount());
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(200)
                        .withBatchIntervalMs(1000)
                        .withMaxRetries(3)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:3306/realtime")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("root")
                        .build()
        );
    }
}