本文主要介紹福布溼在flink實時流處理中，state使用的一些經驗和心得。本文預設圍觀的大神已經對flink有一定了解，如果圍觀過程中發現了有疑問的地方，歡迎在評論區留言。

1. 狀態的類別

1.1 從資料角度看，flink中的狀態分為2種：

1. KeyedState

在按key分割槽的DataStream中，每個key擁有一個自己的state，換句話說，這個state能得到這個key所有的資料。

結合以上的描述，不難得出以下結論，KeyState只能在KeyedStream上使用。

2. 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需要在實現操作符後才行。

2. 原生狀態（Raw State）

由使用者自定義狀態的內部資料結構，靈活度較高。但flink runtime不知道這類狀態內部的資料結構，因此也無法進行相關優化。

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