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
> > >> >
> > >>
> > >
> >
>
