spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Kalin Stoyanov <>
Subject Broadcast size increases with subsequent iterations
Date Fri, 04 Dec 2020 11:59:54 GMT
Hi all,

I have an iterative algorithm in spark that uses each iteration as the
input for the following one, but the size of the data does not change. I am
using localCheckpoint to cut the data's lineage (and also facilitate some
computations that reuse df-s). However, this runs slower and slower as time
goes on, and when I looked at the logs it turned out each job is
broadcasting larger and larger amounts of data. I can't figure out why this
is happening or how to stop it - with the actual size remaining constant
the only thing I can imagine increasing is the lineage data, but that is
cut by the checkpoint...

Here's an abridged version of the script:
#in class 1
while (self.t < self.ttarget):
    newSnapshot, timePassed = self.integrator.advance(self.cluster) #calls
the below function
    self.cluster = newSnapshot
    self.t += timePassed

    if self.dt_out and self.next_out <= self.t:
        self.snapshot() # this saves the dataframe to disk - Job #3 - it
does 1 broadcast
        self.next_out += self.dt_out

#in class 2 - integrator
def advance(self, df_clust):
    df_clust = df_clust.localCheckpoint().repartition(self.nparts, "id") #
Job #1 - does one broadcast
    df_F = self.calc_F(df_clust).localCheckpoint() # Job #2 - does two
    df_v, df_r = self.step_v(df_clust, df_F), self.step_r(
        df_clust, df_F)

    df_clust = df_r.join(df_v, "id")

    return (df_clust, self.dt)

When I checked the logs, as expected they fall into a repeating pattern of
the 3 jobs (I'm saving to disk on every iteration so it's simpler), that
look identical for every iteration. However, the size of ALL broadcasts is
increasing over time - for example

[image: broadcast_1.png]
[image: broadcast_2.png]
I'd really appreciate any insight into what's causing this..


View raw message