You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/08/13 15:47:13 UTC
[GitHub] [spark] baohe-zhang opened a new pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
baohe-zhang opened a new pull request #29425:
URL: https://github.com/apache/spark/pull/29425
### What changes were proposed in this pull request?
This is a follow-up of https://github.com/apache/spark/pull/29149. The change is that when writing a large value list, we divide the value list to a set of 128-values smaller batches, and bulk write these smaller batches one by one. This improvement can reduce memory pressure caused by serialization and give fairness to other writing threads. The idea is proposed by @mridulm in https://github.com/apache/spark/pull/29149#issuecomment-671551444.
### Why are the changes needed?
To reduce the memory pressure and alleviate the fairness issue when batch writing a very long value list.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Manual test.
I measured the HybridStore switching time for different event logs on an HDD disk after the change. It took almost the same amount of time as before.
Log1: 1.3 GB, 1000 jobs, 400 tasks per job
Original: 108s
Now: 107s
Log2: 265 MB, 400 jobs, 200 tasks per job
Original: 23s
Now: 25s
Log3: 133 MB, 400 jobs, 100 tasks per job
Original: 13s
Now: 12s
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-696422422
Hi @HeartSaVioR , I would like to know if there are any updates on the review of this PR. Thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-680293412
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-680361497
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-753551835
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-680361497
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-680293412
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-685170802
**[Test build #128168 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128168/testReport)** for PR 29425 at commit [`5a5f6c9`](https://github.com/apache/spark/commit/5a5f6c9906bfdc5c8244310c921709e2cd6559a9).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-696422422
Hi @HeartSaVioR , I would like to know if there are any updates on the review of this PR. Thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] gatorsmile commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
gatorsmile commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r471163708
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +165,39 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the value list to a set of the 128-values batches. It can reduce the
+ // memory pressure caused by serialization and give fairness to other writing threads
+ // when writing a very large list.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
Review comment:
I left a comment in https://github.com/apache/spark/pull/28412#pullrequestreview-468098043 . I think we should add the unit test cases for verifying the code work as our design. We might need to update the implementation to provide the private APIs for testing.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r479413297
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +165,39 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the value list to a set of the 128-values batches. It can reduce the
+ // memory pressure caused by serialization and give fairness to other writing threads
+ // when writing a very large list.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
+ final Iterator<?> valueIter = batchList.iterator();
+ final Iterator<byte[]> serializedValueIter;
+
+ // Deserialize outside synchronized block
+ List<byte[]> serializedValueList = new ArrayList<>(batchList.size());
+ for (Object value : batchList) {
+ serializedValueList.add(serializer.serialize(value));
+ }
+ serializedValueIter = serializedValueList.iterator();
+
+ final LevelDBTypeInfo ti = getTypeInfo(klass);
+ synchronized (ti) {
+ final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
+ final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
- try (WriteBatch batch = db().createWriteBatch()) {
while (valueIter.hasNext()) {
- updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
- naturalIndex, indices);
+ try (WriteBatch batch = db().createWriteBatch()) {
+ updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
+ naturalIndex, indices);
+ db().write(batch);
Review comment:
Sounds good, I would try to fix it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-685221424
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r476796493
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +165,39 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the value list to a set of the 128-values batches. It can reduce the
+ // memory pressure caused by serialization and give fairness to other writing threads
+ // when writing a very large list.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
Review comment:
Hi @gatorsmile , unit tests for HybridStore are added in https://github.com/apache/spark/pull/29509. And the unit test for **writeAll()** method of levelDB is added in current pr.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-685171231
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] closed pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #29425:
URL: https://github.com/apache/spark/pull/29425
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673631398
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r481479994
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +166,44 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the large value list to a set of smaller batches, to reduce the memory
+ // pressure caused by serialization and give fairness to other writing threads.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
+ final Iterator<?> valueIter = batchList.iterator();
+ final Iterator<byte[]> serializedValueIter;
- try (WriteBatch batch = db().createWriteBatch()) {
- while (valueIter.hasNext()) {
- updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
- naturalIndex, indices);
+ // Deserialize outside synchronized block
+ List<byte[]> serializedValueList = new ArrayList<>(batchList.size());
+ for (Object value : batchList) {
+ serializedValueList.add(serializer.serialize(value));
+ }
+ serializedValueIter = serializedValueList.iterator();
+
+ final LevelDBTypeInfo ti = getTypeInfo(klass);
+ synchronized (ti) {
+ final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
+ final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+
+ try (WriteBatch batch = db().createWriteBatch()) {
+ // A hash map to update the delta of each countKey, wrap countKey with type byte[]
+ // as ByteBuffer because ByteBuffer is comparable.
+ Map<ByteBuffer, Long> counts = new HashMap<>();
+ while (valueIter.hasNext()) {
+ updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
+ naturalIndex, indices, counts);
+ }
+ for (Map.Entry<ByteBuffer, Long> countEntry : counts.entrySet()) {
+ naturalIndex.updateCount(batch, countEntry.getKey().array(), countEntry.getValue());
Review comment:
@HeartSaVioR Here I use naturalIndex.updateCount() to put the count information of all indexes to the batch. When implementing this I found we can lift the method **updateCount()** and **long getCount(byte[] key)** from LevelDBTypeInfo.Index to LevelDBTypeInfo, as these methods are not accessing any member of LevelDBTypeInfo.Index. Doing that would allow us to call ti.updateCount() to update count for all indexes, which would make more sense. However, it's totally optional.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-680292987
**[Test build #127897 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127897/testReport)** for PR 29425 at commit [`ae7d0cb`](https://github.com/apache/spark/commit/ae7d0cb7f2fc88837da0553343890125713501a3).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673557515
**[Test build #127415 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127415/testReport)** for PR 29425 at commit [`17653ba`](https://github.com/apache/spark/commit/17653bab2824d66a38154a6ed1574938c2e3f82f).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673557515
**[Test build #127415 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127415/testReport)** for PR 29425 at commit [`17653ba`](https://github.com/apache/spark/commit/17653bab2824d66a38154a6ed1574938c2e3f82f).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-686342201
I'm not 100% sure I'm qualified to review this patch, as this touches the details of LevelDB KV store implementation.
(My understanding is more likely regarding to SHS + interfaces of KV store, instead of KV store implementation"s".)
Let me cc to @mridulm and @vanzin again. I'll try to spend time to look into details of LevelDB once I get some time and try to review again if @mridulm and @vanzin aren't available for these days.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r478403959
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +165,39 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the value list to a set of the 128-values batches. It can reduce the
+ // memory pressure caused by serialization and give fairness to other writing threads
+ // when writing a very large list.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
+ final Iterator<?> valueIter = batchList.iterator();
+ final Iterator<byte[]> serializedValueIter;
+
+ // Deserialize outside synchronized block
+ List<byte[]> serializedValueList = new ArrayList<>(batchList.size());
+ for (Object value : batchList) {
+ serializedValueList.add(serializer.serialize(value));
+ }
+ serializedValueIter = serializedValueList.iterator();
+
+ final LevelDBTypeInfo ti = getTypeInfo(klass);
+ synchronized (ti) {
+ final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
+ final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
- try (WriteBatch batch = db().createWriteBatch()) {
while (valueIter.hasNext()) {
- updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
- naturalIndex, indices);
+ try (WriteBatch batch = db().createWriteBatch()) {
+ updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
+ naturalIndex, indices);
+ db().write(batch);
Review comment:
Thanks for the information. Sounds like ideally we should fix the bug in LevelDBTypeInfo instead. Sorry I haven't have time to look into details. If possible we'd better fix there; if that's not feasible please let me know.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-680292987
**[Test build #127897 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127897/testReport)** for PR 29425 at commit [`ae7d0cb`](https://github.com/apache/spark/commit/ae7d0cb7f2fc88837da0553343890125713501a3).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-685171231
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673558097
cc @mridulm @HeartSaVioR
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-685170802
**[Test build #128168 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128168/testReport)** for PR 29425 at commit [`5a5f6c9`](https://github.com/apache/spark/commit/5a5f6c9906bfdc5c8244310c921709e2cd6559a9).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-698303116
Sorry I still have several things in my plate and have been struggling with these things. You'd better ping @mridulm as he'd understand the patch well.
@mridulm Appreciated if you have a time to look into this. Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-685220477
**[Test build #128168 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128168/testReport)** for PR 29425 at commit [`5a5f6c9`](https://github.com/apache/spark/commit/5a5f6c9906bfdc5c8244310c921709e2cd6559a9).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r481479994
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +166,44 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the large value list to a set of smaller batches, to reduce the memory
+ // pressure caused by serialization and give fairness to other writing threads.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
+ final Iterator<?> valueIter = batchList.iterator();
+ final Iterator<byte[]> serializedValueIter;
- try (WriteBatch batch = db().createWriteBatch()) {
- while (valueIter.hasNext()) {
- updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
- naturalIndex, indices);
+ // Deserialize outside synchronized block
+ List<byte[]> serializedValueList = new ArrayList<>(batchList.size());
+ for (Object value : batchList) {
+ serializedValueList.add(serializer.serialize(value));
+ }
+ serializedValueIter = serializedValueList.iterator();
+
+ final LevelDBTypeInfo ti = getTypeInfo(klass);
+ synchronized (ti) {
+ final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
+ final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+
+ try (WriteBatch batch = db().createWriteBatch()) {
+ // A hash map to update the delta of each countKey, wrap countKey with type byte[]
+ // as ByteBuffer because ByteBuffer is comparable.
+ Map<ByteBuffer, Long> counts = new HashMap<>();
+ while (valueIter.hasNext()) {
+ updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
+ naturalIndex, indices, counts);
+ }
+ for (Map.Entry<ByteBuffer, Long> countEntry : counts.entrySet()) {
+ naturalIndex.updateCount(batch, countEntry.getKey().array(), countEntry.getValue());
Review comment:
@HeartSaVioR Here I use **naturalIndex.updateCount()** to put the count information of all indexes to the batch. When implementing this I found we can lift the method **updateCount()** and **long getCount(byte[] key)** from LevelDBTypeInfo.Index to LevelDBTypeInfo, as these methods are not accessing any member of LevelDBTypeInfo.Index. Doing that would allow us to call **ti.updateCount()** to update count for all indexes, which would make more sense. However, it's totally optional.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-685221424
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-680359385
**[Test build #127897 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127897/testReport)** for PR 29425 at commit [`ae7d0cb`](https://github.com/apache/spark/commit/ae7d0cb7f2fc88837da0553343890125713501a3).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673558234
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673558234
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-696422422
Hi @HeartSaVioR , I would like to know if there are any updates on the review of this PR. Thanks!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673630432
**[Test build #127415 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127415/testReport)** for PR 29425 at commit [`17653ba`](https://github.com/apache/spark/commit/17653bab2824d66a38154a6ed1574938c2e3f82f).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r471558002
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +165,39 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the value list to a set of the 128-values batches. It can reduce the
+ // memory pressure caused by serialization and give fairness to other writing threads
+ // when writing a very large list.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
Review comment:
Sure I will add unit tests.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-673631398
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #29425:
URL: https://github.com/apache/spark/pull/29425#issuecomment-698303116
Sorry I still have several things in my plate and have been struggling with these things. You'd better ping @mridulm as he'd understand the patch well.
@mridulm Appreciated if you have a time to look into this. Thanks.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Fix count update issue and partition the value list to a set of small batches for LevelDB writeAll
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r481478177
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +165,39 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the value list to a set of the 128-values batches. It can reduce the
+ // memory pressure caused by serialization and give fairness to other writing threads
+ // when writing a very large list.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
+ final Iterator<?> valueIter = batchList.iterator();
+ final Iterator<byte[]> serializedValueIter;
+
+ // Deserialize outside synchronized block
+ List<byte[]> serializedValueList = new ArrayList<>(batchList.size());
+ for (Object value : batchList) {
+ serializedValueList.add(serializer.serialize(value));
+ }
+ serializedValueIter = serializedValueList.iterator();
+
+ final LevelDBTypeInfo ti = getTypeInfo(klass);
+ synchronized (ti) {
+ final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
+ final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
- try (WriteBatch batch = db().createWriteBatch()) {
while (valueIter.hasNext()) {
- updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
- naturalIndex, indices);
+ try (WriteBatch batch = db().createWriteBatch()) {
+ updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
+ naturalIndex, indices);
+ db().write(batch);
Review comment:
Hi @HeartSaVioR , I fixed the bug by using a hashmap to store and update the <countKey, delta> pair in a batch and use the information in that hashmap to update the count right before calling db.write(batch). I updated the test case to cover the count update issue. I also ran benchmark tests for writeAll and put the result in the description section of this PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] baohe-zhang commented on a change in pull request #29425: [SPARK-32350][FOLLOW-UP] Divide the value list into a set of small batches when bulk writing to LevelDB
Posted by GitBox <gi...@apache.org>.
baohe-zhang commented on a change in pull request #29425:
URL: https://github.com/apache/spark/pull/29425#discussion_r476790437
##########
File path: common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDB.java
##########
@@ -164,35 +165,39 @@ public void writeAll(List<?> values) throws Exception {
Preconditions.checkArgument(values != null && !values.isEmpty(),
"Non-empty values required.");
- // Group by class, in case there are values from different classes in the values
+ // Group by class, in case there are values from different classes in the values.
// Typical usecase is for this to be a single class.
// A NullPointerException will be thrown if values contain null object.
for (Map.Entry<? extends Class<?>, ? extends List<?>> entry :
values.stream().collect(Collectors.groupingBy(Object::getClass)).entrySet()) {
-
- final Iterator<?> valueIter = entry.getValue().iterator();
- final Iterator<byte[]> serializedValueIter;
-
- // Deserialize outside synchronized block
- List<byte[]> list = new ArrayList<>(entry.getValue().size());
- for (Object value : values) {
- list.add(serializer.serialize(value));
- }
- serializedValueIter = list.iterator();
-
final Class<?> klass = entry.getKey();
- final LevelDBTypeInfo ti = getTypeInfo(klass);
- synchronized (ti) {
- final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
- final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
+ // Partition the value list to a set of the 128-values batches. It can reduce the
+ // memory pressure caused by serialization and give fairness to other writing threads
+ // when writing a very large list.
+ for (List<?> batchList : Iterables.partition(entry.getValue(), 128)) {
+ final Iterator<?> valueIter = batchList.iterator();
+ final Iterator<byte[]> serializedValueIter;
+
+ // Deserialize outside synchronized block
+ List<byte[]> serializedValueList = new ArrayList<>(batchList.size());
+ for (Object value : batchList) {
+ serializedValueList.add(serializer.serialize(value));
+ }
+ serializedValueIter = serializedValueList.iterator();
+
+ final LevelDBTypeInfo ti = getTypeInfo(klass);
+ synchronized (ti) {
+ final LevelDBTypeInfo.Index naturalIndex = ti.naturalIndex();
+ final Collection<LevelDBTypeInfo.Index> indices = ti.indices();
- try (WriteBatch batch = db().createWriteBatch()) {
while (valueIter.hasNext()) {
- updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
- naturalIndex, indices);
+ try (WriteBatch batch = db().createWriteBatch()) {
+ updateBatch(batch, valueIter.next(), serializedValueIter.next(), klass,
+ naturalIndex, indices);
+ db().write(batch);
Review comment:
@HeartSaVioR Here I put db().write(batch) inside the while loop, that means we will call db().write() for every entity, in a way similar to removeAllByIndexValues(). I did this change because after I added the unit test, I found a bug that the count cannot be updated properly when we use the original code of writeAll(). The count is updated in https://github.com/apache/spark/blob/master/common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDBTypeInfo.java#L395. Since the delta is overridden to 1 every time we add a new entity, we can only increase the count by 1 for every db().write(batch) call, even though we write multiple entities in that call.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org