[
https://issues.apache.org/jira/browse/SPARK-17403?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15984987#comment-15984987
]
Paul Lysak commented on SPARK-17403:
------------------------------------
Hope that helps - finally managed to reproduce it without using production data:
{code}
import org.apache.spark.sql.functions._
val leftRows = spark.sparkContext.parallelize(numSlices = 3000, seq = for
(k <- 1 to 1000; j <- 1 to 1000) yield Row(k, j))
.flatMap(r => (1 to 1000).map(l => Row.fromSeq(r.toSeq :+ l)))
val leftDf = spark.createDataFrame(leftRows, StructType(Seq(
StructField("k", IntegerType),
StructField("j", IntegerType),
StructField("l", IntegerType)
)))
.withColumn("combinedKey", expr("k*1000000 + j*1000 + l"))
.withColumn("fixedCol", lit("sampleVal"))
.withColumn("combKeyStr", format_number(col("combinedKey"), 0))
.withColumn("k100", expr("k*100"))
.withColumn("j100", expr("j*100"))
.withColumn("l100", expr("l*100"))
.withColumn("k_200", expr("k+200"))
.withColumn("j_200", expr("j+200"))
.withColumn("l_200", expr("l+200"))
.withColumn("strCol1_1", concat(lit("value of sample column number one
with which column k will be concatenated:" * 5), format_number(col("k"), 0)))
.withColumn("strCol1_2", concat(lit("value of sample column two one with
which column j will be concatenated:" * 5), format_number(col("j"), 0)))
.withColumn("strCol1_3", concat(lit("value of sample column three one
with which column r will be concatenated:" * 5), format_number(col("l"), 0)))
.withColumn("strCol2_1", concat(lit("value of sample column number one
with which column k will be concatenated:" * 5), format_number(col("k"), 0)))
.withColumn("strCol2_2", concat(lit("value of sample column two one with
which column j will be concatenated:" * 5), format_number(col("j"), 0)))
.withColumn("strCol2_3", concat(lit("value of sample column three one
with which column r will be concatenated:" * 5), format_number(col("l"), 0)))
.withColumn("strCol3_1", concat(lit("value of sample column number one
with which column k will be concatenated:" * 5), format_number(col("k"), 0)))
.withColumn("strCol3_2", concat(lit("value of sample column two one with
which column j will be concatenated:" * 5), format_number(col("j"), 0)))
.withColumn("strCol3_3", concat(lit("value of sample column three one
with which column r will be concatenated:" * 5), format_number(col("l"), 0)))
//if further columns commented out - error disappears
leftDf.cache()
println("===== leftDf count:" + leftDf.count())
leftDf.show(10)
val rightRows = spark.sparkContext.parallelize((1 to 800).map(i => Row(i,
"k_" + i, "sampleVal")))
val rightDf = spark.createDataFrame(rightRows, StructType(Seq(
StructField("k", IntegerType),
StructField("kStr", StringType),
StructField("sampleCol", StringType)
)))
rightDf.cache()
println("===== rightDf count:" + rightDf.count())
rightDf.show(10)
val joinedDf = leftDf.join(broadcast(rightDf), usingColumns = Seq("k"),
joinType = "left")
joinedDf.cache()
println("===== joinedDf count:" + joinedDf.count())
joinedDf.show(10)
{code}
ApplicationMaster fails with such exception:
{noformat}
User class threw exception: org.apache.spark.SparkException: Job aborted due to
stage failure: Task 949 in stage 8.0 failed 4 times, most recent failure: Lost
task 949.3 in stage 8.0 (TID 4922, ip-10-35-162-219.ec2.internal, executor
139): java.lang.NegativeArraySizeException
at org.apache.spark.unsafe.types.UTF8String.getBytes(UTF8String.java:229)
at org.apache.spark.unsafe.types.UTF8String.clone(UTF8String.java:826)
at
org.apache.spark.sql.execution.columnar.StringColumnStats.gatherStats(ColumnStats.scala:216)
at
org.apache.spark.sql.execution.columnar.NullableColumnBuilder$class.appendFrom(NullableColumnBuilder.scala:55)
at
org.apache.spark.sql.execution.columnar.NativeColumnBuilder.org$apache$spark$sql$execution$columnar$compression$CompressibleColumnBuilder$$super$appendFrom(ColumnBuilder.scala:97)
at
org.apache.spark.sql.execution.columnar.compression.CompressibleColumnBuilder$class.appendFrom(CompressibleColumnBuilder.scala:78)
at
org.apache.spark.sql.execution.columnar.NativeColumnBuilder.appendFrom(ColumnBuilder.scala:97)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:122)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation$$anonfun$1$$anon$1.next(InMemoryRelation.scala:97)
at
org.apache.spark.storage.memory.MemoryStore.putIteratorAsValues(MemoryStore.scala:216)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:957)
at
org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:948)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:888)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:948)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:694)
at org.apache.spark.rdd.RDD.getOrCompute(RDD.scala:334)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:285)
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.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.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.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
at org.apache.spark.scheduler.Task.run(Task.scala:99)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
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:745)
{noformat}
We're using Spark 2.1 on YARN (Amazon EMR), tests were performed with executors
running on 15..25 m4.xlarge nodes. Driver ran on r4.xlarge node. Takes 20..30
minutes from start to fail.
Parameters passed to spark-submit:
{noformat}
spark-submit
--deploy-mode cluster
--driver-memory 20700m
--executor-memory 3900m
--conf spark.driver.cores=4
--conf spark.executor.cores=4
--conf spark.sql.broadcastTimeout=1600
--conf spark.network.timeout=1000
--conf spark.scheduler.listenerbus.eventqueue.size=2000000
--conf spark.dynamicAllocation.maxExecutors=200
--conf spark.dynamicAllocation.enabled=true
--conf spark.yarn.maxAppAttempts=1
--conf spark.yarn.driver.memoryOverhead=2600
--conf spark.yarn.executor.memoryOverhead=1700
--conf spark.sql.shuffle.partitions=3000
...files, jar, class, program args etc. go next...
{noformat}
Timeouts and queue size parameters are essential to reproduce the error -
without them error changes. As I said in previous comment, it's very sensitive
to exact data and parameters.
Unfortunately, I wasn't able to identify in more details where lies the
boundary between successful run and such error as launch sometimes takes quite
a long time.
> Fatal Error: Scan cached strings
> --------------------------------
>
> Key: SPARK-17403
> URL: https://issues.apache.org/jira/browse/SPARK-17403
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 2.0.0
> Environment: Spark standalone cluster (3 Workers, 47 cores)
> Ubuntu 14
> Java 8
> Reporter: Ruben Hernando
>
> The process creates views from JDBC (SQL server) source and combines them to
> create other views.
> Finally it dumps results via JDBC
> Error:
> {quote}
> # JRE version: Java(TM) SE Runtime Environment (8.0_101-b13) (build
> 1.8.0_101-b13)
> # Java VM: Java HotSpot(TM) 64-Bit Server VM (25.101-b13 mixed mode
> linux-amd64 )
> # Problematic frame:
> # J 4895 C1 org.apache.spark.unsafe.Platform.getLong(Ljava/lang/Object;J)J (9
> bytes) @ 0x00007fbb355dfd6c [0x00007fbb355dfd60+0xc]
> #
> {quote}
> SQL Query plan (fields truncated):
> {noformat}
> == Parsed Logical Plan ==
> 'Project [*]
> +- 'UnresolvedRelation `COEQ_63`
> == Analyzed Logical Plan ==
> InstanceId: bigint, price: double, ZoneId: int, priceItemId: int, priceId: int
> Project [InstanceId#20236L, price#20237, ZoneId#20239, priceItemId#20242,
> priceId#20244]
> +- SubqueryAlias coeq_63
> +- Project [_TableSL_SID#143L AS InstanceId#20236L, SL_RD_ColR_N#189 AS
> price#20237, 24 AS ZoneId#20239, 6 AS priceItemId#20242, 63 AS priceId#20244]
> +- SubqueryAlias 6__input
> +-
> Relation[_TableSL_SID#143L,_TableP_DC_SID#144L,_TableSH_SID#145L,ID#146,Name#147,TableP_DCID#148,TableSHID#149,SL_ACT_GI_DTE#150,SL_Xcl_C#151,SL_Xcl_C#152,SL_Css_Cojs#153L,SL_Config#154,SL_CREATEDON#
> .......... 36 more fields] JDBCRelation((select [SLTables].[_TableSL_SID],
> [SLTables]. ... [...] FROM [sch].[SLTables] [SLTables] JOIN sch.TPSLTables
> TPSLTables ON [TPSLTables].[_TableSL_SID] = [SLTables].[_TableSL_SID] where
> _TP = 24) input)
> == Optimized Logical Plan ==
> Project [_TableSL_SID#143L AS InstanceId#20236L, SL_RD_ColR_N#189 AS
> price#20237, 24 AS ZoneId#20239, 6 AS priceItemId#20242, 63 AS priceId#20244]
> +- InMemoryRelation [_TableSL_SID#143L, _TableP_DC_SID#144L,
> _TableSH_SID#145L, ID#146, Name#147, ... 36 more fields], true, 10000,
> StorageLevel(disk, memory, deserialized, 1 replicas)
> : +- *Scan JDBCRelation((select [SLTables].[_TableSL_SID],
> [SLTables].[_TableP_DC_SID], [SLTables].[_TableSH_SID], [SLTables].[ID],
> [SLTables].[Name], [SLTables].[TableP_DCID], [SLTables].[TableSHID],
> [TPSLTables].[SL_ACT_GI_DTE], ... [...] FROM [sch].[SLTables] [SLTables]
> JOIN sch.TPSLTables TPSLTables ON [TPSLTables].[_TableSL_SID] =
> [SLTables].[_TableSL_SID] where _TP = 24) input)
> [_TableSL_SID#143L,_TableP_DC_SID#144L,_TableSH_SID#145L,ID#146,Name#147,TableP_DCID#148,TableSHID#149,SL_ACT_GI_DTE#150,SL_Xcl_C#151,...
> 36 more fields]
> == Physical Plan ==
> *Project [_TableSL_SID#143L AS InstanceId#20236L, SL_RD_ColR_N#189 AS
> price#20237, 24 AS ZoneId#20239, 6 AS priceItemId#20242, 63 AS priceId#20244]
> +- InMemoryTableScan [_TableSL_SID#143L, SL_RD_ColR_N#189]
> : +- InMemoryRelation [_TableSL_SID#143L, _TableP_DC_SID#144L,
> _TableSH_SID#145L, ID#146, Name#147, ... 36 more fields], true, 10000,
> StorageLevel(disk, memory, deserialized, 1 replicas)
> : : +- *Scan JDBCRelation((select [SLTables].[_TableSL_SID],
> [SLTables].[_TableP_DC_SID], [SLTables].[_TableSH_SID], [SLTables].[ID],
> [SLTables].[Name], [SLTables].[TableP_DCID], ... [...] FROM [sch].[SLTables]
> [SLTables] JOIN sch.TPSLTables TPSLTables ON [TPSLTables].[_TableSL_SID] =
> [SLTables].[_TableSL_SID] where _TP = 24) input)
> [_TableSL_SID#143L,_TableP_DC_SID#144L,_TableSH_SID#145L,ID#146,Name#147,,...
> 36 more fields]
> {noformat}
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
