Flink状态妙用

本文主要介绍福布湿在flink实时流处理中，state使用的一些经验和心得。本文默认围观的大神已经对flink有一定了解，如果围观过程中发现了有疑问的地方，欢迎在评论区留言。

1. 状态的类别

1.1 从数据角度看，flink中的状态分为2种：

  1. KeyedState

在按key分区的DataStream中，每个key拥有一个自己的state，换句话说，这个state能得到这个key所有的数据。

结合以上的描述，不难得出以下结论，KeyState只能在KeyedStream上使用。

  1. OperateState

OperateState得到的数据是当前算子实例接收到的数据，换句话说，有几个算子实例就有几个对应的OperateState。

1.2 从flink

runtime 对状态支持的机制不同也分为2种：

  1. 托管状态（Managed State）

flink runtime知道这类状态的内部数据结构，在状态进行保存和更新或者dataStream并行度发生改变以及内存管理方面flink runtime能对过程进行优化，提升效率。这类状态是官方推荐。

更重要的是，所有的DataStream function（map、filter、apply等其他所有操作函数）均支持managed state，但是raw state需要在实现操作符后才行。

  1. 原生状态（Raw State）

由用户自定义状态的内部数据结构，灵活度较高。但flink runtime不知道这类状态内部的数据结构，因此也无法进行相关优化。

Managed State Raw State
KeyedState ValueState < T > -
ListState < T >
MapState<UK,UV>
ReducingState < T >
AggregatingState<IN, OUT>
OperateState CheckpointedFunction -
ListCheckpointed < T extends Serializable >

1.3 案例-不同店铺累计商品销售额排行

1.3.1 Scala版本

import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.windowing.time.Time

// 这行引用十分重要，许多隐式转换以及Flink SQL中的列表达式等均包含在此引用中
import org.apache.flink.streaming.api.scala._

object StateExample {

  case class Order(finishTime: Long, memberId: Long, productId: Long, sale: Double)

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(5000)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    /**
     *设置状态存储方式，一般有以下几种存储方式：
     *          类名               存储位置                                  一般使用环境
     * 1  MemoryStateBackend   内存中                                    是用本地调试，或者是状态很小的情况
     * 2  FsStateBackend       落地到文件系统，堆内存会缓存正在传输的数据    适用生产环境，满足HA，性能大于3小于1，但不支持增量更新
     *                         有OOM风险
     * 3  RocksDBStateBackend  落地到文件系统，RocksDB数据库在本地磁盘上    适用生产环境（建议使用此项），满足HA，支持增量更新
     *                         缓存传输中的数据
     **/
    env.setStateBackend(new RocksDBStateBackend("oss://bigdata/xxx/order-state"))

    val dataStream: DataStream[Order] = env
      .fromCollection((1 to 25)
      .map(i => Order(i, i % 7, i % 3, i + 0.1)))
      /**
       * 自定义事件时间
       **/
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Order](Time.milliseconds(1)) {
        override def extractTimestamp(element: Order): Long = element.finishTime
      })


    //实时输出 不同店铺累计商品销售额排行
    dataStream.keyBy("memberId")
      .mapWithState[List[Order],List[Order]] {
        case (order: Order, None) => (order +: Nil,Some(List(order)))
        case (order: Order, Some(orders:List[Order])) => {
          val l = (orders :+ order).groupBy(_.productId).mapValues {
            case List(o) => o
            case l: List[Order] => l.reduce((a, b) => Order(if (a.finishTime > b.finishTime) a.finishTime else b.finishTime, a.memberId, a.productId, a.sale + b.sale))
          }.values.toList.sortWith(_.sale > _.sale)
          (l,Some(l))
        }
      }.print()

    env.execute("example")
  }
}

1.3.2 java版本

import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;

public class StateExampleJ {
    static final SimpleDateFormat YYYY_MM_DD_HH = new SimpleDateFormat("yyyyMMdd HH");

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //env.setStateBackend(new RocksDBStateBackend("oss://bigdata/xxx/order-state"));
        List<Order> data = new LinkedList<>();
        for (long i = 1; i <= 25; i++)
            data.add(new Order(i, i % 7, i % 3, i + 0.1));
        DataStream<Order> dataStream = env.fromCollection(data).setParallelism(1).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Order>(Time.milliseconds(1)) {
            @Override
            public long extractTimestamp(Order element) {
                return element.finishTime;
            }
        });
        dataStream.keyBy(o -> o.memberId).map(
                new RichMapFunction<Order, List<Order>>() {
                    MapState<Long, Order> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        MapStateDescriptor<Long, Order> productRank = new MapStateDescriptor<Long, Order>("productRank", Long.class, Order.class);
                        mapState = getRuntimeContext().getMapState(productRank);
                    }

                    @Override
                    public List<Order> map(Order value) throws Exception {
                        if (mapState.contains(value.productId)) {
                            Order acc = mapState.get(value.productId);
                            value.sale += acc.sale;
                        }
                        mapState.put(value.productId, value);
                        return IteratorUtils.toList(mapState.values().iterator());
                    }
                }
        ).print();


        env.execute("exsample");
    }


    public static class Order {
        //finishTime: Long, memberId: Long, productId: Long, sale: Double
        public long finishTime;
        public long memberId;
        public long productId;
        public double sale;

        public Order() {
        }

        public Order(Long finishTime, Long memberId, Long productId, Double sale) {
            this.finishTime = finishTime;
            this.memberId = memberId;
            this.productId = productId;
            this.sale = sale;
        }
    }
}

2. 状态针对迟到数据的优化

实时处理面对的第一个难题就是迟到事件的处理（或者说是流乱序的处理）。想必各位同学都有被迟到事件折磨过的经验。虽然官方API提供了迟到数据处理的机制：

(1) assignTimestampsAndWatermarks

.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Order>(Time.milliseconds(1)) {
    @Override
    public long extractTimestamp(Order element) {
        return element.finishTime;
    }
});

(2) allowedLateness

.timeWindow(Time.days(1)).allowedLateness(Time.seconds(1)).sideOutputLateData(outputTag)

但是我想说，这2个迟到时间设太小满足不了精度要求，设太大又会导致性能问题，然后你就会拿历史数据分析计算合适的迟到时间，然后你会发现特么运气不好的时候依然会出现过大的误差。福布湿在这里给大家提供一种解决迟到问题的一种思路，废话不多说，直接上代码，关于其中的一些说明和解释福布湿在代码中已注释的形式说明。

代码框架沿用1.3.2

主要处理逻辑

static final SimpleDateFormat YYYY_MM_DD_HH = new SimpleDateFormat("yyyyMMdd HH");
// 实时输出每个小时每个店铺商品的排行
dataStream
        .keyBy(o -> o.memberId)
        .map(new RichMapFunction<Order, MemberRank>() {
            MapState<String, MemberRank> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.hours(1)) //设置状态的超时时间为1个小时
                        //设置ttl更新策略为创建和写，直观作用为如果一个key（例如20200101 01）1个小时内没有写入的操作，只有读的操作，那么这个key将被标记为超时
                        //值得注意的是，MapState ListState这类集合state，超时机制作用在每个元素上，也就是每个元素的超时是独立的
                        .updateTtlOnCreateAndWrite()
                        .cleanupInBackground() // 指定过期的key清除的操作策略
                        .build();
                MapStateDescriptor<String, MemberRank> descriptor = new MapStateDescriptor<String, MemberRank>("hourRank", String.class, MemberRank.class);
                descriptor.enableTimeToLive(ttlConfig);
                mapState = getRuntimeContext().getMapState(descriptor);
            }

            @Override
            public MemberRank map(Order value) throws Exception {
                String key = YYYY_MM_DD_HH.format(value.finishTime);
                MemberRank rank;
                if (mapState.contains(key)) {
                    rank = mapState.get(key);
                    rank.merge(value);

                } else {
                    rank = MemberRank.of(value);
                }
                mapState.put(key, rank);
                return rank;
            }
        }).print();

内部类MemberRank

public static class MemberRank {
    public String time;
    public long memberId;
    public List<Order> rank;

    public MemberRank() {
    }

    public MemberRank(String time, long memberId, List<Order> rank) {
        this.time = time;
        this.memberId = memberId;
        this.rank = rank;
    }

    public static MemberRank of(Order o) {
        return new MemberRank(YYYY_MM_DD_HH.format(o.finishTime), o.memberId, Collections.singletonList(o));
    }

    public void merge(Order o) {
        rank.forEach(e -> {
            if (e.productId == o.productId) {
                e.sale += o.sale;
            }
        });
        rank.sort((o1, o2) -> Double.valueOf((o1.sale - o2.sale) * 1000).intValue());
    }
}

3. 基于状态的维表关联

维表关联，flink已经有了很好很成熟的接口，福布湿用过的有：

（1） AsyncDataStream.unorderedWait()

（2） Join

（3） BroadcastStream

这几个各有特点，AsyncDataStream.unorderedWait效率最高，但是需要源支持异步客户端，join维表方面个人用的比较少，BroadcastStream没有什么特殊限制，性能也还行，算是比较通用，但是不能定期更新维表信息。

也许你想到了，当源不支持异步客户端，而维表数据又更新的相对较为频繁的时候，以上方式好像都不太适合，下面福布湿就把自己的一些经验介绍给大家。

废话不多说，直接上代码。

import com.fulu.stream.source.http.SyncHttpClient;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.configuration.Configuration;

public class MainOrderHttpMap extends RichMapFunction<SimpleOrder, SimpleOrder> {
    transient MapState<String, Member> member;
    transient SyncHttpClient client;
    public MainOrderHttpMap() {}
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        StateTtlConfig updateTtl = StateTtlConfig
                .newBuilder(org.apache.flink.api.common.time.Time.days(1))
                .updateTtlOnCreateAndWrite()
                .neverReturnExpired()
                .build();


        MapStateDescriptor<String, Member> memberDesc = new MapStateDescriptor<String, Member>("member-map", String.class, Member.class);
        memberDesc.enableTimeToLive(updateTtl);
        member = getRuntimeContext().getMapState(memberDesc);
        
    }

    @Override
    public SimpleOrder map(SimpleOrder value) throws Exception {
        value.profitCenterName = getProfitCenter(value.memberId);
        return value;
    }
    
    private String getProfitCenter(String id) throws Exception {
        String name = null;
        int retry = 1;
        while (name == null &amp;&amp; retry <= 3) {
            if (member.contains(id))
                name = member.get(id).profitCenterName;
            else {
                Member m = client.queryMember(id);
                if (m != null) {
                    member.put(id, m);
                    name = m.profitCenterName;
                }
            }
            retry++;
        }
        return name;
    }
    
    @Override
    public void close() throws Exception {
        super.close();
        client.close();
    }
}

想必各位同学直接就能看懂，是的原理很简单，就是将维表缓存在状态中，同时制定状态的过期时间以达到定期更新的目的。

4. Distinct语义

细心的同学可能已经发现，DataStream类中没有distinct Operation。但是当源中存在少量重复数据时怎么办呢，没错，使用状态缓存所有的事件id ，然后使用filter进行过滤操作，由于原理确实很简单，福布湿就不贴代码了。

5. 结尾

福布湿在实时流处理方面最先接触的是spark-streaming，因此在初期学习flink时感觉最难啃的就是state这一块，因此在这里特地将福布斯关于状态的一些经验分享给大家。相信大家在熟悉state后会彻底爱上flink。

参考文献：

【1】 Flink官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.11/concepts/stateful-stream-processing.html

【2】 https://www.jianshu.com/p/ac0fff780d40?from=singlemessage

【3】 https://zhuanlan.zhihu.com/p/136722111

福禄ICH·大数据开发团队 福布湿
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/fulu/p/13410794.html