Mailing-List: contact dev-help@spark.apache.org; run by ezmlm
Precedence: bulk
MIME-Version: 1.0
In-Reply-To: <1459441588178-16944.post@n3.nabble.com>
References: <1459441588178-16944.post@n3.nabble.com>
From: Michael Armbrust <michael@databricks.com>
Date: Fri, 1 Apr 2016 13:29:11 -0700
Message-ID: 
 <CAAswR-6EpTFpjsn8goofxmAx5Lfyb7xba7X4cCPx2EWv403Q3Q@mail.gmail.com>
Subject: Re: What influences the space complexity of Spark operations?
To: Steve Johnston <sjohnston@algebraixdata.com>
Cc: "dev@spark.apache.org" <dev@spark.apache.org>
Content-Type: multipart/alternative; boundary=001a11401cb2822dc7052f723c69

--001a11401cb2822dc7052f723c69
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

Blocking operators like Sort, Join or Aggregate will put all of the data
for a whole partition into a hash table or array.  However, if you are
running Spark 1.5+ we should be spilling to disk.  In Spark 1.6 if you are
seeing OOMs for SQL operations you should report it as a bug.

On Thu, Mar 31, 2016 at 9:26 AM, Steve Johnston <sjohnston@algebraixdata.co=
m
> wrote:

> *What we=E2=80=99ve observed*
>
> Increasing the number of partitions (and thus decreasing the partition
> size) seems to reliably help avoid OOM errors. To demonstrate this we use=
d
> a single executor and loaded a small table into a DataFrame, persisted it
> with MEMORY_AND_DISK, repartitioned it and joined it to itself. Varying t=
he
> number of partitions identifies a threshold between completing the join a=
nd
> incurring an OOM error.
>
>
> lineitem =3D sc.textFile('lineitem.tbl').map(converter)
> lineitem =3D sqlContext.createDataFrame(lineitem, schema)
> lineitem.persist(StorageLevel.MEMORY_AND_DISK)
> repartitioned =3D lineitem.repartition(partition_count)
> joined =3D repartitioned.join(repartitioned)
> joined.show()
>
>
> *Questions*
>
> Generally, what influences the space complexity of Spark operations? Is i=
t
> the case that a single partition of each operand=E2=80=99s data set + a s=
ingle
> partition of the resulting data set all need to fit in memory at the same
> time? We can see where the transformations (for say joins) are implemente=
d
> in the source code (for the example above BroadcastNestedLoopJoin), but
> they seem to be based on virtualized iterators; where in the code is the
> partition data for the inputs and outputs actually materialized?
> ------------------------------
> View this message in context: What influences the space complexity of
> Spark operations?
> <http://apache-spark-developers-list.1001551.n3.nabble.com/What-influence=
s-the-space-complexity-of-Spark-operations-tp16944.html>
> Sent from the Apache Spark Developers List mailing list archive
> <http://apache-spark-developers-list.1001551.n3.nabble.com/> at
> Nabble.com.
>
>

--001a11401cb2822dc7052f723c69
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">Blocking operators like Sort, Join or Aggregate will put a=
ll of the data for a whole partition into a hash table or array.=C2=A0 Howe=
ver, if you are running Spark 1.5+ we should be spilling to disk.=C2=A0 In =
Spark 1.6 if you are seeing OOMs for SQL operations you should report it as=
 a bug.</div><div class=3D"gmail_extra"><br><div class=3D"gmail_quote">On T=
hu, Mar 31, 2016 at 9:26 AM, Steve Johnston <span dir=3D"ltr">&lt;<a href=
=3D"mailto:sjohnston@algebraixdata.com" target=3D"_blank">sjohnston@algebra=
ixdata.com</a>&gt;</span> wrote:<br><blockquote class=3D"gmail_quote" style=
=3D"margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex"><b>What =
we=E2=80=99ve observed</b>
<p>
Increasing the number of partitions (and thus decreasing the partition size=
) seems to reliably help avoid OOM errors. To demonstrate this we used a si=
ngle executor and loaded a small table into a DataFrame, persisted it with =
MEMORY_AND_DISK, repartitioned it and joined it to itself. Varying the numb=
er of partitions identifies a threshold between completing the join and inc=
urring an OOM error.=20

</p><pre><code>
lineitem =3D sc.textFile(&#39;lineitem.tbl&#39;).map(converter)
lineitem =3D sqlContext.createDataFrame(lineitem, schema)
lineitem.persist(StorageLevel.MEMORY_AND_DISK)
repartitioned =3D lineitem.repartition(partition_count)
joined =3D repartitioned.join(repartitioned)
joined.show()
 </code></pre>

<b>Questions</b><p> Generally, what influences the space complexity of Spar=
k operations? Is it the case that a single partition of each operand=E2=80=
=99s data set + a single partition of the resulting data set all need to fi=
t in memory at the same time? We can see where the transformations (for say=
 joins) are implemented in the source code (for the example above Broadcast=
NestedLoopJoin), but they seem to be based on virtualized iterators; where =
in the code is the partition data for the inputs and outputs actually mater=
ialized?

=09
=09
=09
<br></p><hr align=3D"left" width=3D"300">
View this message in context: <a href=3D"http://apache-spark-developers-lis=
t.1001551.n3.nabble.com/What-influences-the-space-complexity-of-Spark-opera=
tions-tp16944.html" target=3D"_blank">What influences the space complexity =
of Spark operations?</a><br>
Sent from the <a href=3D"http://apache-spark-developers-list.1001551.n3.nab=
ble.com/" target=3D"_blank">Apache Spark Developers List mailing list archi=
ve</a> at Nabble.com.<br><p></p><p></p></blockquote></div><br></div>

--001a11401cb2822dc7052f723c69--