Spark: Parallelize initializing readTasks when localityPreferred is true #2800
Conversation
| .stopOnFailure() | ||
| .executeWith(readTasksInitExecutorService) | ||
| .run(task -> { | ||
| readTasks.add(new ReadTask<>( |
There was a problem hiding this comment.
This can't alter readTasks from multiple threads because array lists aren't thread-safe.
There was a problem hiding this comment.
Oh, that was a mistake. Fixed it!
|
Oops, I have also fix this a few days ago, and create another PR #2803 just a moment ago. |
| task, tableBroadcast, expectedSchemaString, caseSensitive, | ||
| localityPreferred, InternalRowReaderFactory.INSTANCE); | ||
| synchronized (readTasks) { | ||
| readTasks.add(readTask); |
There was a problem hiding this comment.
Maybe it‘s better to use array instead of list, and then we can avoid to add lock or temporary var ?
There was a problem hiding this comment.
Yes, using array can help avoid lock, but the lock time should be insignificant compared to the get block location operation. The return value is a list and I didn't want to do all the transformation.
There was a problem hiding this comment.
Yeap! But another reason is we can keep the style consistent with spark3 module, a little bit of optimization with array as follow :R153, how do you think?
| InputPartition<ColumnarBatch> readTask = new ReadTask<>( | ||
| task, tableBroadcast, expectedSchemaString, caseSensitive, | ||
| localityPreferred, new BatchReaderFactory(batchSize)); | ||
| synchronized (readTasks) { |
There was a problem hiding this comment.
Why not use a concurrent list?
There was a problem hiding this comment.
At first time I wanted to use a concurrent list. But this would change the return value from a normal list to a concurrent list, and the subsequent operation on this list should only be read. Maybe it will affect performance.
There was a problem hiding this comment.
I changed the code to use arrays like in spark3 to avoid synchronization.
jshmchenxi
force-pushed
the
multi-readtask-init
branch
from
7000ecc to
1313fa9
Compare
| Tasks.range(readTasks.length) | ||
| .stopOnFailure() | ||
| .executeWith(readTasksInitExecutorService) | ||
| .run(index -> { |
There was a problem hiding this comment.
The block is unnecessary because there is only one expression, can you remove it?
| private Filter[] pushedFilters = NO_FILTERS; | ||
| private final boolean localityPreferred; | ||
| private final int batchSize; | ||
| private ExecutorService readTasksInitExecutorService = DEFAULT_READTASKS_INIT_EXECUTOR_SERVICE; |
There was a problem hiding this comment.
I don't think that there is a need for this field and I'd prefer not to add mutable state. Can you refactor this to call executeWith(ThreadPools.getWorkerPool()) instead? You can pass null to that method so it can also check localityPreferred inline:
.executeWith(localityPreferred ? ThreadPools.getWorkerPool() : null)
There was a problem hiding this comment.
I like this comment because it really simplify the code path without introducing any static or local variables.
There was a problem hiding this comment.
That really simplifies the code. Thanks @rdblue
|
Thanks for @jshmchenxi for the optimize work for spark on HDFS ( I just notice this and #2577, it's impressive). Do we also need to parallelize the SparkMicroBatchStream#planInputPartitions in this PR ? |
|
@openinx Thanks for reminding, I've added parallelization to SparkMicroBatchStream#planInputPartitions in this PR. |
|
Thanks, @jshmchenxi! I merged this. |
4 participants
Follows #2577, use thread pool to initialize readTasks if Spark locality is preferred.
Before this, the Spark plan phase could be slow as it uses single thread to obtain block locations of all files for this scan.
More information could be found in this comment