ok thanks for sharing - I'll have a look later this week.
Regards,
Matthias
On Mon, May 8, 2017 at 2:20 PM, Mingyang Wang <miw...@eng.ucsd.edu> wrote:
> Hi Matthias,
>
> With a driver memory of 10GB, all operations were executed on CP, and I did
> observe that the version of reading FK as a vector and then converting it
> was faster, which took 8.337s (6.246s on GC) while the version of reading
> FK as a matrix took 31.680s (26.256s on GC).
>
> For the distributed caching, I have re-run all scripts with the following
> Spark configuration
>
> --driver-memory 1G \
> --executor-memory 100G \
> --executor-cores 20 \
> --num-executors 1 \
> --conf spark.driver.maxResultSize=0 \
> --conf spark.rpc.message.maxSize=128 \
>
> And it seems that both versions have some problems.
>
> 1) Sum of FK in matrix form
> ```
> FK = read($FK)
> print("Sum of FK = " + sum(FK))
> ```
> Worked as expected. Took 8.786s.
>
>
> 2) Sum of FK in matrix form, with checkpoints
> ```
> FK = read($FK)
> if (1 == 1) {}
> print("Sum of FK = " + sum(FK))
> ```
> It took 89.731s, with detailed stats shown below.
>
> 17/05/08 13:15:00 INFO api.ScriptExecutorUtils: SystemML Statistics:
> Total elapsed time: 91.619 sec.
> Total compilation time: 1.889 sec.
> Total execution time: 89.731 sec.
> Number of compiled Spark inst: 2.
> Number of executed Spark inst: 2.
> Cache hits (Mem, WB, FS, HDFS): 1/0/0/0.
> Cache writes (WB, FS, HDFS): 0/0/0.
> Cache times (ACQr/m, RLS, EXP): 0.000/0.001/0.000/0.000 sec.
> HOP DAGs recompiled (PRED, SB): 0/0.
> HOP DAGs recompile time: 0.000 sec.
> Spark ctx create time (lazy): 0.895 sec.
> Spark trans counts (par,bc,col):0/0/0.
> Spark trans times (par,bc,col): 0.000/0.000/0.000 secs.
> Total JIT compile time: 5.001 sec.
> Total JVM GC count: 8.
> Total JVM GC time: 0.161 sec.
> Heavy hitter instructions (name, time, count):
> -- 1) sp_uak+ 89.349 sec 1
> -- 2) sp_chkpoint 0.381 sec 1
> -- 3) == 0.001 sec 1
> -- 4) + 0.000 sec 1
> -- 5) print 0.000 sec 1
> -- 6) castdts 0.000 sec 1
> -- 7) createvar 0.000 sec 3
> -- 8) rmvar 0.000 sec 7
> -- 9) assignvar 0.000 sec 1
> -- 10) cpvar 0.000 sec 1
>
>
> 3) Sum of FK in vector form
> ```
> FK_colvec = read($FK_colvec)
> FK = table(seq(1,nrow(FK_colvec)), FK_colvec, nrow(FK_colvec), 1e6)
> print("Sum of FK = " + sum(FK))
> ```
> Things really went wrong. It took ~10 mins.
>
> 17/05/08 13:26:36 INFO api.ScriptExecutorUtils: SystemML Statistics:
> Total elapsed time: 605.688 sec.
> Total compilation time: 1.857 sec.
> Total execution time: 603.832 sec.
> Number of compiled Spark inst: 2.
> Number of executed Spark inst: 2.
> Cache hits (Mem, WB, FS, HDFS): 1/0/0/0.
> Cache writes (WB, FS, HDFS): 0/0/0.
> Cache times (ACQr/m, RLS, EXP): 0.000/0.000/0.000/0.000 sec.
> HOP DAGs recompiled (PRED, SB): 0/1.
> HOP DAGs recompile time: 0.002 sec.
> Spark ctx create time (lazy): 0.858 sec.
> Spark trans counts (par,bc,col):0/0/0.
> Spark trans times (par,bc,col): 0.000/0.000/0.000 secs.
> Total JIT compile time: 3.682 sec.
> Total JVM GC count: 5.
> Total JVM GC time: 0.064 sec.
> Heavy hitter instructions (name, time, count):
> -- 1) sp_uak+ 603.447 sec 1
> -- 2) sp_rexpand 0.381 sec 1
> -- 3) createvar 0.002 sec 3
> -- 4) rmvar 0.000 sec 5
> -- 5) + 0.000 sec 1
> -- 6) print 0.000 sec 1
> -- 7) castdts 0.000 sec 1
>
> Also, from the executor log, there were some disk spilling:
>
> 17/05/08 13:20:00 INFO ExternalSorter: Thread 109 spilling in-memory
> map of 33.8 GB to disk (1 time so far)
> 17/05/08 13:20:20 INFO ExternalSorter: Thread 116 spilling in-memory
> map of 31.2 GB to disk (1 time so far)
>
> ...
>
> 17/05/08 13:24:50 INFO ExternalAppendOnlyMap: Thread 116 spilling
> in-memory map of 26.9 GB to disk (1 time so far)
> 17/05/08 13:25:08 INFO ExternalAppendOnlyMap: Thread 109 spilling
> in-memory map of 26.6 GB to disk (1 time so far)
>
>
>
> Regards,
> Mingyang
>
> On Sat, May 6, 2017 at 9:12 PM Matthias Boehm <mboe...@googlemail.com>
> wrote:
>
> > yes, even with the previous patch for improved memory efficiency of
> > ultra-sparse matrices in MCSR format, there is still some unnecessary
> > overhead that leads to garbage collection. For this reason, I would
> > recommend to read it as vector and convert it in memory to an
> ultra-sparse
> > matrix. I also just pushed a minor performance improvement for reading
> > ultra-sparse matrices but the major bottleneck still exist.
> >
> > The core issue is that we can't read these ultra-sparse matrices into a
> CSR
> > representation because it does not allow for efficient incremental
> > construction (with unordered inputs and multi-threaded read). However, I
> > created SYSTEMML-1587 to solve this in the general case. The idea is to
> > read ultra-sparse matrices into thread-local COO deltas and finally merge
> > it into a CSR representation. The initial results are very promising and
> > it's safe because the temporary memory requirements are covered by the
> MCSR
> > estimate, but it will take a while because I want to introduce this
> > consistently for all readers (single-/multi-threaded, all formats).
> >
> > In contrast to the read issue, I was not able to reproduce the described
> > performance issue of distributed caching. Could you please double check
> > that this test also used the current master build and perhaps share the
> > detailed setup again (e.g., num executors, data distribution, etc).
> Thanks.
> >
> > Regards,
> > Matthias
> >
> >
> > On Thu, May 4, 2017 at 9:55 PM, Mingyang Wang <miw...@eng.ucsd.edu>
> wrote:
> >
> > > Out of curiosity, I increased the driver memory to 10GB, and then all
> > > operations were executed on CP. It took 37.166s but JVM GC took
> 30.534s.
> > I
> > > was wondering whether this is the expected behavior?
> > >
> > > Total elapsed time: 38.093 sec.
> > > Total compilation time: 0.926 sec.
> > > Total execution time: 37.166 sec.
> > > Number of compiled Spark inst: 0.
> > > Number of executed Spark inst: 0.
> > > Cache hits (Mem, WB, FS, HDFS): 0/0/0/1.
> > > Cache writes (WB, FS, HDFS): 0/0/0.
> > > Cache times (ACQr/m, RLS, EXP): 30.400/0.000/0.001/0.000 sec.
> > > HOP DAGs recompiled (PRED, SB): 0/0.
> > > HOP DAGs recompile time: 0.000 sec.
> > > Spark ctx create time (lazy): 0.000 sec.
> > > Spark trans counts (par,bc,col):0/0/0.
> > > Spark trans times (par,bc,col): 0.000/0.000/0.000 secs.
> > > Total JIT compile time: 22.302 sec.
> > > Total JVM GC count: 11.
> > > Total JVM GC time: 30.534 sec.
> > > Heavy hitter instructions (name, time, count):
> > > -- 1) uak+ 37.166 sec 1
> > > -- 2) == 0.001 sec 1
> > > -- 3) + 0.000 sec 1
> > > -- 4) print 0.000 sec 1
> > > -- 5) rmvar 0.000 sec 5
> > > -- 6) createvar 0.000 sec 1
> > > -- 7) assignvar 0.000 sec 1
> > > -- 8) cpvar 0.000 sec 1
> > >
> > > Regards,
> > > Mingyang
> > >
> > > On Thu, May 4, 2017 at 9:48 PM Mingyang Wang <miw...@eng.ucsd.edu>
> > wrote:
> > >
> > > > Hi Matthias,
> > > >
> > > > Thanks for the patch.
> > > >
> > > > I have re-run the experiment and observed that there was indeed no
> more
> > > > memory pressure, but it still took ~90s for this simple script. I was
> > > > wondering what is the bottleneck for this case?
> > > >
> > > >
> > > > Total elapsed time: 94.800 sec.
> > > > Total compilation time: 1.826 sec.
> > > > Total execution time: 92.974 sec.
> > > > Number of compiled Spark inst: 2.
> > > > Number of executed Spark inst: 2.
> > > > Cache hits (Mem, WB, FS, HDFS): 1/0/0/0.
> > > > Cache writes (WB, FS, HDFS): 0/0/0.
> > > > Cache times (ACQr/m, RLS, EXP): 0.000/0.000/0.000/0.000 sec.
> > > > HOP DAGs recompiled (PRED, SB): 0/0.
> > > > HOP DAGs recompile time: 0.000 sec.
> > > > Spark ctx create time (lazy): 0.860 sec.
> > > > Spark trans counts (par,bc,col):0/0/0.
> > > > Spark trans times (par,bc,col): 0.000/0.000/0.000 secs.
> > > > Total JIT compile time: 3.498 sec.
> > > > Total JVM GC count: 5.
> > > > Total JVM GC time: 0.064 sec.
> > > > Heavy hitter instructions (name, time, count):
> > > > -- 1) sp_uak+ 92.597 sec 1
> > > > -- 2) sp_chkpoint 0.377 sec 1
> > > > -- 3) == 0.001 sec 1
> > > > -- 4) print 0.000 sec 1
> > > > -- 5) + 0.000 sec 1
> > > > -- 6) castdts 0.000 sec 1
> > > > -- 7) createvar 0.000 sec 3
> > > > -- 8) rmvar 0.000 sec 7
> > > > -- 9) assignvar 0.000 sec 1
> > > > -- 10) cpvar 0.000 sec 1
> > > >
> > > > Regards,
> > > > Mingyang
> > > >
> > > > On Wed, May 3, 2017 at 8:54 AM Matthias Boehm <
> mboe...@googlemail.com>
> > > > wrote:
> > > >
> > > >> to summarize, this was an issue of selecting serialized
> > representations
> > > >> for large ultra-sparse matrices. Thanks again for sharing your
> > feedback
> > > >> with us.
> > > >>
> > > >> 1) In-memory representation: In CSR every non-zero will require 12
> > bytes
> > > >> - this is 240MB in your case. The overall memory consumption,
> however,
> > > >> depends on the distribution of non-zeros: In CSR, each block with at
> > > >> least one non-zero requires 4KB for row pointers. Assuming uniform
> > > >> distribution (the worst case), this gives us 80GB. This is likely
> the
> > > >> problem here. Every empty block would have an overhead of 44Bytes
> but
> > > >> for the worst-case assumption, there are no empty blocks left. We do
> > not
> > > >> use COO for checkpoints because it would slow down subsequent
> > > operations.
> > > >>
> > > >> 2) Serialized/on-disk representation: For sparse datasets that are
> > > >> expected to exceed aggregate memory, we used to use a serialized
> > > >> representation (with storage level MEM_AND_DISK_SER) which uses
> > sparse,
> > > >> ultra-sparse, or empty representations. In this form, ultra-sparse
> > > >> blocks require 9 + 16*nnz bytes and empty blocks require 9 bytes.
> > > >> Therefore, with this representation selected, you're dataset should
> > > >> easily fit in aggregate memory. Also, note that chkpoint is only a
> > > >> transformation that persists the rdd, the subsequent operation then
> > > >> pulls the data into memory.
> > > >>
> > > >> At a high-level this was a bug. We missed ultra-sparse
> representations
> > > >> when introducing an improvement that stores sparse matrices in MCSR
> > > >> format in CSR format on checkpoints which eliminated the need to
> use a
> > > >> serialized storage level. I just deliver a fix. Now we store such
> > > >> ultra-sparse matrices again in serialized form which should
> > > >> significantly reduce the memory pressure.
> > > >>
> > > >> Regards,
> > > >> Matthias
> > > >>
> > > >> On 5/3/2017 9:38 AM, Mingyang Wang wrote:
> > > >> > Hi all,
> > > >> >
> > > >> > I was playing with a super sparse matrix FK, 2e7 by 1e6, with only
> > one
> > > >> > non-zero value on each row, that is 2e7 non-zero values in total.
> > > >> >
> > > >> > With driver memory of 1GB and executor memory of 100GB, I found
> the
> > > HOP
> > > >> > "Spark chkpoint", which is used to pin the FK matrix in memory, is
> > > >> really
> > > >> > expensive, as it invokes lots of disk operations.
> > > >> >
> > > >> > FK is stored in binary format with 24 blocks, each block is ~45MB,
> > and
> > > >> ~1GB
> > > >> > in total.
> > > >> >
> > > >> > For example, with the script as
> > > >> >
> > > >> > """
> > > >> > FK = read($FK)
> > > >> > print("Sum of FK = " + sum(FK))
> > > >> > """
> > > >> >
> > > >> > things worked fine, and it took ~8s.
> > > >> >
> > > >> > While with the script as
> > > >> >
> > > >> > """
> > > >> > FK = read($FK)
> > > >> > if (1 == 1) {}
> > > >> > print("Sum of FK = " + sum(FK))
> > > >> > """
> > > >> >
> > > >> > things changed. It took ~92s and I observed lots of disk spills
> from
> > > >> logs.
> > > >> > Based on the stats from Spark UI, it seems the materialized FK
> > > requires
> > > >> >> 54GB storage and thus introduces disk operations.
> > > >> >
> > > >> > I was wondering, is this the expected behavior of a super sparse
> > > matrix?
> > > >> >
> > > >> >
> > > >> > Regards,
> > > >> > Mingyang
> > > >> >
> > > >>
> > > >
> > >
> >
>
