Spark sql "create table as select ..." outofmemory

0
votes

I am using Spark SQL. Spark version 2.2.2.

The SQL likes

select a.*, b.* from a inner join b on a.id=b.id

Table a and b is huge. This works and I can get the answer.

Secondly, I want to save the SQL results to hive table. The SQL likes:

create table xxx stored as orc as select ...

This would fail with error code 143. The error message is:

ExecutorLostFailure (executor 268 exited caused by one of the running tasks) Reason: Container marked as failed: container_e37_1554167308087_15187_01_000269 on host: xxx. Exit status: 143. Diagnostics: Container killed on request. Exit code is 143 +details ExecutorLostFailure (executor 268 exited caused by one of the running tasks) Reason: Container marked as failed: container_e37_1554167308087_15187_01_000269 on host: xxxx. Exit status: 143. Diagnostics: Container killed on request. Exit code is 143 Container exited with a non-zero exit code 143 Killed by external signal

I tried to use pyspark to use df.saveAsTable(). This works at some scenarios but not all times.

Can anybody help?

Thanks.

1
apache-sparkapache-spark-sql
Hive creates a lot of meta data handled by the driver. How much RAM did you allocate the driver --driver-memory? - MaFF
df.saveAsTable() fails randomly because of uneven partitionning. try repartitioning before saving df.repartition(n).saveAsTable() - MaFF
@MaFF Not fails randomly. It works for table a and b but fails for table c and d. - Longxing Wei
dirver-memory is 6GB @MaFF - Longxing Wei
to solve memory issues you can try repartitioning, increase executor memory --executor-memory, reduce number of cores by executor (they share the RAM of an executor), don't forget to increase the memory overhead to match the specifications - MaFF

1 Answers

1
votes

To solve memory issues you can try :

  • repartitioning
  • increase executor memory --executor-memory
  • reduce number of cores by executor --executor-cores (they share the RAM of an executor), don't forget to increase the memory overhead to match the specifications

Repartitioning, will reduce the size of each task while increasing memory or reducing the number of cores will result in more memory allocated to each task.

If you don't know how many tasks you'll need beforehand and how much memory you should allocate, you can repartition according to the size of your dataframes. 

df.persist()
n = df.count()
nb_records = 10
df = df.repartition(int(n / nb_records))
df.rdd.getNumPartitions()

10

Here we make sure to have 10 records by task, the dataframe has 100 records therefore I asked for 10 partitions.