火花流：为什么处理几MB的用户状态的内部处理成本如此之高？

Question

根据我们的实验，我们发现当状态变成超过一百万个对象时，有状态的Spark Streaming内部处理成本会花费大量时间。因此，延迟会受到影响，因为我们必须增加批处理间隔以避免不稳定的行为（处理时间

它与我们应用程序的细节无关，因为它可以通过下面的代码复制。

Spark内部处理/基础架构成本到底是什么，需要这么多时间来处理用户状态？除了简单地增加批处理间隔之外，还有什么选项可以减少流转时长吗？

我们计划广泛使用状态：在几个节点中的每一个节点上至少 100MB 左右，将所有数据保存在内存中，并且每小时只转储一次。

增加批处理间隔有帮助，但我们希望保持批处理间隔最小。

原因可能不是状态占用的空间，而是大对象图，因为当我们将列表更改为大基元数组时，问题就消失了。

只是一个猜测：它可能与Spark内部使用的org.apache.spark.util.SizeEstimator有关，因为它不时在分析时出现。

这是一个简单的演示，可以在现代iCore7上重现上面的图片：

• 小于15 MB的状态
• 根本没有流输入
• 最快可能（虚拟）'updateStateByKey'函数
• 批处理间隔1秒
• 检查点（Spark需要，必须有）到本地磁盘
• 在本地和纱线上进行了测试

代码：

package spark;

import org.apache.commons.lang3.RandomStringUtils;
import org.apache.spark.HashPartitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.util.SizeEstimator;
import scala.Tuple2;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

public class SlowSparkStreamingUpdateStateDemo {

    // Very simple state model
    static class State implements Serializable {
        final List<String> data;
        State(List<String> data) {
            this.data = data;
        }
    }

    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
                // Tried KryoSerializer, but it does not seem to help much
                //.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .setMaster("local[*]")
                .setAppName(SlowSparkStreamingUpdateStateDemo.class.getName());

        JavaStreamingContext javaStreamingContext = new JavaStreamingContext(conf, Durations.seconds(1));
        javaStreamingContext.checkpoint("checkpoint"); // a must (if you have stateful operation)

        List<Tuple2<String, State>> initialRddGeneratedData = prepareInitialRddData();
        System.out.println("Estimated size, bytes: " + SizeEstimator.estimate(initialRddGeneratedData));
        JavaPairRDD<String, State> initialRdd = javaStreamingContext.sparkContext().parallelizePairs(initialRddGeneratedData);

        JavaPairDStream<String, State> stream = javaStreamingContext
                .textFileStream(".") // fake: effectively, no input at all
                .mapToPair(input -> (Tuple2<String, State>) null) //  fake to get JavaPairDStream
                .updateStateByKey(
                        (inputs, maybeState) -> maybeState, // simplest possible dummy function
                        new HashPartitioner(javaStreamingContext.sparkContext().defaultParallelism()),
                        initialRdd); // set generated state

        stream.foreachRDD(rdd -> { // simplest possible action (required by Spark)
            System.out.println("Is empty: " + rdd.isEmpty());
            return null;
        });

        javaStreamingContext.start();
        javaStreamingContext.awaitTermination();
    }

    private static List<Tuple2<String, State>> prepareInitialRddData() {
        // 'stateCount' tuples with value = list of size 'dataListSize' of strings of length 'elementDataSize'
        int stateCount = 1000;
        int dataListSize = 200;
        int elementDataSize = 10;
        List<Tuple2<String, State>> initialRddInput = new ArrayList<>(stateCount);
        for (int stateIdx = 0; stateIdx < stateCount; stateIdx++) {
            List<String> stateData = new ArrayList<>(dataListSize);
            for (int dataIdx = 0; dataIdx < dataListSize; dataIdx++) {
                stateData.add(RandomStringUtils.randomAlphanumeric(elementDataSize));
            }
            initialRddInput.add(new Tuple2<>("state" + stateIdx, new State(stateData)));
        }
        return initialRddInput;
    }

}

Answer 1

Spark 1.6中改进了状态管理。
请参阅SPARK-2629改进的Spark Streams状态管理；

在详细设计规范中：
改进了 Spark 流中的状态管理

一个性能缺陷如下：