Spark SQL 1.6.0 - massive memory usage for simple query

Question

I'm using Spark 1.6 on EMR 4.3 to query ~15TB of data belonging to a table in a hive metastore (backed by gzipped parquet files in S3). For my cluster I have a r3.8xlarge master node and 15 r3.8xlarge core nodes (3.6TB RAM, 9.6TB SSD).

The ~15TB data is contained in maybe 9 billion rows. Each row has ~15 columns that store strings of length 5-50, and one column which contains an array of ~30 strings, 10-20 characters each. There are only ~1 million unique strings stored in the array. All I'm trying to do is count the unique strings in the array column, but it seems I'm running out of memory as I keep getting: OutOfMemoryError: unable to create new native thread on the executors. The tasks fail due to the out of memory error, the executors get disabled, and then the job fails.

It works when I query 5-10TB of data. I must not understand correctly what gets stored in memory (which is what I'm trying to figure out). Btw, with the above cluster, I'm setting:

spark.executor.memory 30g
spark.executor.cores 5
spark.executor.instances 90 // 6 instances per r3.8xlarge host

I didn't think Spark SQL stored the intermediary tables in memory. Since there aren't more than 1M unique strings, I would think that the strings with their counts should easily fit in memory. Here's the query:

val initial_df = sqlContext.sql("select unique_strings_col from Table where timestamp_partition between '2016-09-20T07:00:00Z' and '2016-09-23T07:00:00Z'")
initial_df.registerTempTable("initial_table") // ~15TB compressed data to read in from S3

val unique_strings_df = sqlContext.sql("select posexplode(unique_strings_col) as (string_pos, string) from initial_table").select($"string_pos", $"string")
unique_strings_df.registerTempTable("unique_strings_table")  // ~70% initial data remaining at this point

val strings_count_df = sqlContext.sql("select string, count(*) as unique_string_count from unique_strings_table where string_pos < 21 group by string order by unique_string_count desc") // ~50% initial data remaining at this point
strings_count_df.write.parquet("s3://mybucket/counts/2016-09-20-2016-09-23")

The compressed parquet files are small (say 5mb each). It seems like they could be read one at a time, filtered, and stored with their counts. What am I missing?

Answer 1

So it turns out that I need to have enough disk + memory space to store the initial RDD. If I do more up-front filtering in the initial RDD before creating the temp table, I'm able to run the query successfully. Yay!