You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "SteNicholas (via GitHub)" <gi...@apache.org> on 2023/08/08 13:45:08 UTC
[GitHub] [hudi] SteNicholas opened a new pull request, #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
SteNicholas opened a new pull request, #9395:
URL: https://github.com/apache/hudi/pull/9395
### Change Logs
`HoodieEngineContext` should not use parallel stream with parallelism greater than the number of CPU cores to avoid `OutOfMemoryError` of `ForkJoinTask`, of which stacktrace as follows:
```
Caused by: java.lang.OutOfMemoryError
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:598)
at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677)
at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735)
at java.util.stream.ReduceOps$ReduceOp.evaluateParallel(ReduceOps.java:714)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
at org.apache.hudi.client.common.HoodieFlinkEngineContext.map(HoodieFlinkEngineContext.java:101)
at org.apache.hudi.table.action.clean.CleanPlanActionExecutor.requestClean(CleanPlanActionExecutor.java:117)
at org.apache.hudi.table.action.clean.CleanPlanActionExecutor.requestClean(CleanPlanActionExecutor.java:145)
at org.apache.hudi.table.action.clean.CleanPlanActionExecutor.execute(CleanPlanActionExecutor.java:170)
at org.apache.hudi.table.HoodieFlinkCopyOnWriteTable.scheduleCleaning(HoodieFlinkCopyOnWriteTable.java:353)
at org.apache.hudi.client.BaseHoodieWriteClient.scheduleTableServiceInternal(BaseHoodieWriteClient.java:1434)
at org.apache.hudi.client.BaseHoodieWriteClient.clean(BaseHoodieWriteClient.java:891)
at org.apache.hudi.async.AsyncCleanerService.lambda$startService$0(AsyncCleanerService.java:68)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)
```
### Impact
`HoodieEngineContext` does not use `BaseStream#parallel()` with parallelism greater than CPU cores.
### Risk level (write none, low medium or high below)
none.
### Documentation Update
none.
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Change Logs and Impact were stated clearly
- [x] Adequate tests were added if applicable
- [x] CI passed
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1328024397
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieEngineContext.java:
##########
@@ -105,4 +105,12 @@ public abstract <I, K, V> List<V> reduceByKey(
public abstract void cancelJob(String jobId);
public abstract void cancelAllJobs();
+
+ public <T> Stream<T> stream(List<T> data, Integer parallelism) {
+ return stream(stream(data.stream(), data.size()), parallelism);
+ }
+
+ public <T> Stream<T> stream(Stream<T> data, Integer parallelism) {
+ return parallelism == null || parallelism > Runtime.getRuntime().availableProcessors() ? data : data.parallel();
Review Comment:
Got it. @stream2000 do you still have concerns?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] stream2000 commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "stream2000 (via GitHub)" <gi...@apache.org>.
stream2000 commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1288537503
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
> @stream2000, the parallelism of `stream().parallel()` is only `Runtime.getRuntime().availableProcessors()`
So actually the parallelism of `availableProcessors` will cause OOM? If I configure the parallelism just as `Runtime.getRuntime().availableProcessors()` we still get OOM right? Correct me if I'm wrong~
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1670603500
@danny0405, the failure of CI is fixed by #9401.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] stream2000 commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "stream2000 (via GitHub)" <gi...@apache.org>.
stream2000 commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1288371818
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
When the parallelism is 1000, will we run it in 1000 parallelism or `Runtime.getRuntime().availableProcessors()`? If we use just the parallelism as `Runtime.getRuntime().availableProcessors()`, will it cause `OutOfMemoryError` since it it not actually a large parallelism?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1288493882
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
@stream2000, the parallelism of `stream().parallel()` is only `Runtime.getRuntime().availableProcessors()`, then when the parallelism is 1000, it will cause `OutOfMemoryError` by using `stream().parallel()`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1287825581
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
What is the default parallelism for `stream().parallel()` ?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1288493882
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
@stream2000, the parallelism of `stream().parallel()` is only `Runtime.getRuntime().availableProcessors()`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1671137882
@stream2000, thanks for the fix. I have rebased the lastest master branch. cc @danny0405.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] danny0405 commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1329449151
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieEngineContext.java:
##########
@@ -105,4 +105,12 @@ public abstract <I, K, V> List<V> reduceByKey(
public abstract void cancelJob(String jobId);
public abstract void cancelAllJobs();
+
+ public <T> Stream<T> stream(List<T> data, Integer parallelism) {
+ return stream(stream(data.stream(), data.size()), parallelism);
+ }
+
+ public <T> Stream<T> stream(Stream<T> data, Integer parallelism) {
+ return parallelism == null || parallelism > Runtime.getRuntime().availableProcessors() ? data : data.parallel();
Review Comment:
agree with @stream2000 , we should not limit the parallelism for all cases.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1326787156
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieEngineContext.java:
##########
@@ -105,4 +105,12 @@ public abstract <I, K, V> List<V> reduceByKey(
public abstract void cancelJob(String jobId);
public abstract void cancelAllJobs();
+
+ public <T> Stream<T> stream(List<T> data, Integer parallelism) {
+ return stream(stream(data.stream(), data.size()), parallelism);
+ }
+
+ public <T> Stream<T> stream(Stream<T> data, Integer parallelism) {
+ return parallelism == null || parallelism > Runtime.getRuntime().availableProcessors() ? data : data.parallel();
Review Comment:
@yihua, I have tested this pull request and the OOM frequency is reduced. The point of the pull request is aimed to reduce the OOM frequency for using parallelism stream.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1669677534
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1669967315
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213",
"triggerID" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1326688166
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieEngineContext.java:
##########
@@ -105,4 +105,12 @@ public abstract <I, K, V> List<V> reduceByKey(
public abstract void cancelJob(String jobId);
public abstract void cancelAllJobs();
+
+ public <T> Stream<T> stream(List<T> data, Integer parallelism) {
+ return stream(stream(data.stream(), data.size()), parallelism);
+ }
+
+ public <T> Stream<T> stream(Stream<T> data, Integer parallelism) {
+ return parallelism == null || parallelism > Runtime.getRuntime().availableProcessors() ? data : data.parallel();
Review Comment:
So if `parallelism > Runtime.getRuntime().availableProcessors()`, the `data` is not parallelized at all, as Java's `data.stream()` is sequential by default. Also to @stream2000 's point, before this PR, the parallelism of `map` is already limited by `Runtime.getRuntime().availableProcessors()`. Does that still cause OOM? Is OOM from somewhere else?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1671580470
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213",
"triggerID" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"triggerType" : "PUSH"
}, {
"hash" : "f20fe8b171dc78a61639c1eabd7c5e5b4bbac201",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19227",
"triggerID" : "f20fe8b171dc78a61639c1eabd7c5e5b4bbac201",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f20fe8b171dc78a61639c1eabd7c5e5b4bbac201 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19227)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1669694218
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213",
"triggerID" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] stream2000 commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "stream2000 (via GitHub)" <gi...@apache.org>.
stream2000 commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1288224103
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
Hi @SteNicholas, correct me if I'm wrong. Do we just run the map function sequentially when the parallelism is larger than the default parallelism? And why there is an `OutOfMemoryError ` risk when we use `stream().parallel()` which will submit some future tasks to the default thread pool which may not cost a lot of memory?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1288326850
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
@stream2000, I don't think only run the map function sequentially. In order to reduce the OOM risk caused by parallelization, all functions should be handled in this way. For example, when the parallelism is 1000, the `CleanPlanActionExecutor` uses map function and may casue `OutOfMemoryError` for many filegroups in `stream().parallel()`, which stacktrace like above description. BTW, the parallel stream only improves a little performance, therefore this change doesn't destroy much performance improvement.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1671200936
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213",
"triggerID" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"triggerType" : "PUSH"
}, {
"hash" : "f20fe8b171dc78a61639c1eabd7c5e5b4bbac201",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f20fe8b171dc78a61639c1eabd7c5e5b4bbac201",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213)
* f20fe8b171dc78a61639c1eabd7c5e5b4bbac201 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] stream2000 commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "stream2000 (via GitHub)" <gi...@apache.org>.
stream2000 commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1670984818
@SteNicholas Hi, sorry for the failure ci introduced. Now we can rebase the lastest master and test
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1326688613
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieEngineContext.java:
##########
@@ -105,4 +105,12 @@ public abstract <I, K, V> List<V> reduceByKey(
public abstract void cancelJob(String jobId);
public abstract void cancelAllJobs();
+
+ public <T> Stream<T> stream(List<T> data, Integer parallelism) {
+ return stream(stream(data.stream(), data.size()), parallelism);
+ }
+
+ public <T> Stream<T> stream(Stream<T> data, Integer parallelism) {
+ return parallelism == null || parallelism > Runtime.getRuntime().availableProcessors() ? data : data.parallel();
Review Comment:
@SteNicholas have you tested this PR that with this change only, the OOM goes away?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] stream2000 commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "stream2000 (via GitHub)" <gi...@apache.org>.
stream2000 commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1328657021
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieEngineContext.java:
##########
@@ -105,4 +105,12 @@ public abstract <I, K, V> List<V> reduceByKey(
public abstract void cancelJob(String jobId);
public abstract void cancelAllJobs();
+
+ public <T> Stream<T> stream(List<T> data, Integer parallelism) {
+ return stream(stream(data.stream(), data.size()), parallelism);
+ }
+
+ public <T> Stream<T> stream(Stream<T> data, Integer parallelism) {
+ return parallelism == null || parallelism > Runtime.getRuntime().availableProcessors() ? data : data.parallel();
Review Comment:
@yihua I agree that reducing the parallelism can avoid OOM in some situations. I'm just wondering should we reduce the parallelism automatically. `HoodieFlinkEngineContext.map` provides a param `parallelism` and we can configure the parallelism as we want. In this scenario (Clean OOM in flink Engine) we can just set `hoodie.cleaner.parallelism` as 1 to reduce the parallelism.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1287885445
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
@danny0405, the default parallelism is `Runtime.getRuntime().availableProcessors()`. `BaseStream#parallel()` uses the `ForkJoinTask` mode.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1287885445
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
@danny0405, the default parallelism is `Runtime.getRuntime().availableProcessors()`. `BaseStream#parallel()` uses the `ForkJoinTask` mode. Meanwhile, if the parallelism is greater than `Runtime.getRuntime().availableProcessors()`, there is `OutOfMemoryError` risk for `BaseStream#parallel()`. Therefore, when the parallelism is greater than default parallelism, we could not use `BaseStream#parallel()` .
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1288326850
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
@stream2000, I don't think only run the map function sequentially. In order to reduce the OOM risk caused by parallelization, all functions should be handled in this way. BTW, for example, when the parallelism is 1000, the `CleanPlanActionExecutor` uses map function and may casue `OutOfMemoryError` for many filegroups in `stream().parallel()`, which stacktrace like above description.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] SteNicholas commented on a diff in pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #9395:
URL: https://github.com/apache/hudi/pull/9395#discussion_r1287885445
##########
hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/common/HoodieFlinkEngineContext.java:
##########
@@ -102,12 +102,12 @@ public RuntimeContext getRuntimeContext() {
@Override
public <I, O> List<O> map(List<I> data, SerializableFunction<I, O> func, int parallelism) {
- return data.stream().parallel().map(throwingMapWrapper(func)).collect(Collectors.toList());
+ return stream(data, parallelism).map(throwingMapWrapper(func)).collect(Collectors.toList());
}
Review Comment:
@danny0405, the default parallelism is `Runtime.getRuntime().availableProcessors()`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #9395:
URL: https://github.com/apache/hudi/pull/9395#issuecomment-1671265476
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213",
"triggerID" : "a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc",
"triggerType" : "PUSH"
}, {
"hash" : "f20fe8b171dc78a61639c1eabd7c5e5b4bbac201",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19227",
"triggerID" : "f20fe8b171dc78a61639c1eabd7c5e5b4bbac201",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a60f7f89b5377119bf8bef6c7ddfd0dc821de1fc Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19213)
* f20fe8b171dc78a61639c1eabd7c5e5b4bbac201 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=19227)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores [hudi]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #9395: [HUDI-6669] HoodieEngineContext should not use parallel stream with parallelism greater than CPU cores
URL: https://github.com/apache/hudi/pull/9395
--
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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org