apache-spark - Spark结构化流的内存问题

Question

我在运行Spark 2.2.0中具有聚合和分区的结构化流时遇到内存问题:

session
    .readStream()
    .schema(inputSchema)
    .option(OPTION_KEY_DELIMITER, OPTION_VALUE_DELIMITER_TAB)
    .option(OPTION_KEY_QUOTE, OPTION_VALUE_QUOTATION_OFF)
    .csv("s3://test-bucket/input")
    .as(Encoders.bean(TestRecord.class))
    .flatMap(mf, Encoders.bean(TestRecord.class))
    .dropDuplicates("testId", "testName")
    .withColumn("year", functions.date_format(dataset.col("testTimestamp").cast(DataTypes.DateType), "YYYY"))
    .writeStream()
    .option("path", "s3://test-bucket/output")
    .option("checkpointLocation", "s3://test-bucket/checkpoint")
    .trigger(Trigger.ProcessingTime(60, TimeUnit.SECONDS))
    .partitionBy("year")
    .format("parquet")
    .outputMode(OutputMode.Append())
    .queryName("test-stream")
    .start();

在测试期间，我注意到每次有新数据出现时，使用的内存量都会增加，最终执行程序以代码137退出:

ExecutorLostFailure (executor 2 exited caused by one of the running tasks) Reason: Container marked as failed: container_1520214726510_0001_01_000003 on host: ip-10-0-1-153.us-west-2.compute.internal. Exit status: 137. Diagnostics: Container killed on request. Exit code is 137
Container exited with a non-zero exit code 137
Killed by external signal

我创建了一个堆转储，发现 StateStore引用了 org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider使用的大部分内存

乍一看看起来很正常，因为这是Spark将聚合 key 保存在内存中的方式。但是，我通过重命名源文件夹中的文件进行了测试，以便可以通过spark拾取它们。由于输入记录是相同的，因此所有其他行都应作为重复项被拒绝，并且内存消耗不应增加，而是应该增加。



此外，GC时间占用了总处理时间的30％以上



这是从执行程序获取的堆转储，其执行器的内存比上面的屏幕要少，因为当我从该执行器创建转储时，java进程刚刚在该过程的中间终止。

Answer 1

迁移我对SPARK-23682的评论，该问题的询问者也已提交。

在HDFS状态存储提供程序中，它过多地将状态的多个版本缓存在内存中，默认为100个版本。 SPARK-24717解决了该问题，它将仅在内存中维护两个版本的状态(用于重放的当前版本，用于更新的新版本)。该补丁将在Spark 2.4.0中提供。