Why does collecting dataset fail with org.apache.spark.shuffle.FetchFailedException?

user_8091789 0
Viewed 4131

I use Spark with YARN cluster manager.

I create a dataset from a Cassandra table with around 700 rows with 5 columns in that and one of the column with data in JSON format. The amount of data is in MB only.

I run spark-shell with:

  • spark.executor.memory=4g
  • spark.driver.memory=2g

I am getting this error:

org.apache.spark.shuffle.FetchFailedException: Failed to connect to bosirahtaicas02.bscuat.local/172.17.0.1:53093 at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:323)

while I am trying to collect data from my dataFrame

I run the following piece of code directly in spark-shell (line by line):

import org.apache.spark.sql._
import org.apache.spark.sql.cassandra._

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkContext

import com.datastax.spark.connector.cql._
import com.datastax.spark.connector._

import org.json._
import java.sql.Timestamp

val sqc = new SQLContext(sc)
val csc = new CassandraSQLContext(sc)

val allRecordsDF = sqc.read.format("org.apache.spark.sql.cassandra").
    option("table", "xyz").
    option("keyspace", 'pqr").load().cache()

val allRowsDF = allRecordsDF.select(allRecordsDF.col("record_type"), allRecordsDF.col("process_time"))

val allPatientsTS = allRowsDF.where(allRowsDF("record_type") === "patient").select("process_time").sort("process_time")

Here when I am trying to collect allPatientsTS Dataframe it is showing error.

spark-cassandra-connector apache-spark-sql apache-spark cassandra scala
2 Answers

I have faced this similar issue. To find a solution I have spent a few weeks to implement suggestions by many sources those are: configuration, partitioning, memory fraction increment, network timeout and so on. However, all failed to provide me with a solution.

Finally, I figured out the issue in my small cluster. The workers node's RAM size is 8 GB, and I have configured the worker memory 6 GB (for larger overhead) in .conf file. Unfortunately, a background service in one of the workers is consuming 3 GB memory and that's why the issue is occurring (as all the workers are seeking to map 6GB memory for each). I downgraded to 4GB worker memory in the spark-defaults.conf file, and now I am not facing this issue. N.B: check all the workers whether any other service is consuming the significant amount of resource.

user_2984230
Related