Gerard Maas created SPARK-21710:
-----------------------------------
Summary: ConsoleSink causes OOM crashes with large inputs.
Key: SPARK-21710
URL: https://issues.apache.org/jira/browse/SPARK-21710
Project: Spark
Issue Type: Bug
Components: Structured Streaming
Affects Versions: 2.2.0
Environment: affects all environments
Reporter: Gerard Maas
ConsoleSink does a full collect of the streaming dataset in order to show few
lines on screen. This is problematic with large inputs, like a kafka backlog or
a file source with files larger than the driver's memory.
Here's an example:
{code:scala}
import spark.implicits._
import org.apache.spark.sql.functions
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types._
val schema = StructType(StructField("text", StringType, true) :: Nil)
val lines = spark
.readStream
.format("text")
.option("path", "/tmp/data")
.schema(schema)
.load()
val base = lines.writeStream
.outputMode("append")
.format("console")
.start()
{code}
When a large file larger than the available driver memory is fed through this
streaming job, we get:
{code:java}
-------------------------------------------
Batch: 0
-------------------------------------------
[Stage 0:> (0 + 8) /
111]17/08/11 15:10:45 ERROR Executor: Exception in task 6.0 in stage 0.0 (TID 6)
java.lang.OutOfMemoryError: Java heap space
at java.util.Arrays.copyOf(Arrays.java:3236)
at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:118)
at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:153)
at
net.jpountz.lz4.LZ4BlockOutputStream.flushBufferedData(LZ4BlockOutputStream.java:205)
at net.jpountz.lz4.LZ4BlockOutputStream.write(LZ4BlockOutputStream.java:158)
at java.io.DataOutputStream.write(DataOutputStream.java:107)
at
org.apache.spark.sql.catalyst.expressions.UnsafeRow.writeToStream(UnsafeRow.java:554)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:237)
at
org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:228)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
at
org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:108)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
17/08/11 15:10:45 ERROR SparkUncaughtExceptionHandler: Uncaught exception in
thread Thread[Executor task launch worker for task 6,5,main]
java.lang.OutOfMemoryError: Java heap space
{code}
This issue can be traced back to a `collect` on the source `DataFrame`:
https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/console.scala#L52
A fairly simple solution would be to do a `take(numRows)` instead of the
collect. (PR in progress)
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
