apache / apache/arrow

[C++][HDFS] Can't get performance improve when increase the thread number of IO thread pool

Open
#33,627 7 comments 0 reactions 0 assignees View on GitHub
Component: C++ Type: enhancement
Dominant language
C++
Stars
17.1k
Forks
4.3k
Avg merge
3d 13h
Merged PRs (30d)
88

Description

### Describe the enhancement requested

Hi, I've tested an ExecPlan and the read time is almost 80% of the total time(though I can't give out the exact read time, I've printed out the log in the HDFS read interface, and it shows read ended at almost 80% of total).
And here is my test results, it tells me the increase of ARROW_IO_THREADS doesn't help to improve performance.
1. total time: 322.502s(OMP_NUM_THREADS=10 ARROW_IO_THREADS=1)
2. total time: 321.271s(OMP_NUM_THREADS=10 ARROW_IO_THREADS=2)
3. total time: 325.166s(OMP_NUM_THREADS=10 ARROW_IO_THREADS=3)
when I read the code I found no matter how many IO thread Arrow used, **it will only use one HDFS connection**, I draw a thread model I learned from the code(pls point out if I'm wrong).
![thread model](https://user-images.githubusercontent.com/10483852/211994276-a4d54d91-c165-46bb-9f18-2bb059210b87.png)

Also, I've checked the CPU usage when I run the tests, the CPU usage of the read threads is very low, about 15%--22% each. I guess the CPU rate can't increase is because of the output IO upper bound of one HDFS connection.(OMP_NUM_THREADS=10 ARROW_IO_THREADS=3)
![CPU usage](https://user-images.githubusercontent.com/10483852/212008263-a29500b4-e829-4364-a807-91d02dcea93f.jpg)

here are some pieces of my code
```
int main() {
arrow::dataset::internal::Initialize();

// default context will only use one thread
ExecContext execContext;
ExecContext exec_context(arrow::default_memory_pool(), ::arrow::internal::GetCpuThreadPool());

auto&& resultPlan = ExecPlan::Make(&exec_context);
if (!resultPlan.ok()) {
WIND_ERR_LOG("Executor run task fail because of make exec plan failed. detail:{}", resultPlan.status().ToString());
return wind::api::ActionResult{ErrorCode::ERR_COLUMN_ENGINE_INIT};
}
std::shared_ptr plan = std::move(resultPlan).ValueUnsafe();
std::vector decls;
auto actionNode = api::CreateDataFrame(task_.plan, dataFrameMap_, schemas_, *config_, outputPath, decls);

Declaration::Sequence(decls).AddToPlan(plan.get());

if (!plan->Validate().ok()) {
WIND_ERR_LOG("Executor run task fail because of can't get validate exec plan. detail:{}", plan->ToString());
return wind::api::ActionResult{ErrorCode::ERR_COLUMN_ENGINE_CREATE_EXEC_PLAN};
}

plan->StartProducing();
auto future = plan->finished();

future.Wait();
}

std::shared_ptr MakeDataSet() {
wind::column::MakeFileSystem(schema_.file.path);

auto format = wind::column::MakeFileFormat(schema_.file.option.csvOption.delimiters[0]);

auto schema = MakeSchemaAndUpdateFileFormat(format);
if (!schema) {
WIND_ERR_LOG("MakeDataSet fail, detail: make schema failed.");
return nullptr;
}

auto fs = wind::column::GetFileSystem();
if (!fs) {
WIND_ERR_LOG("MakeDataSet fail, detail: make file system failed.");
return nullptr;
}

auto&& dataset = ds::FileSystemDataset::Make(schema, cp::literal(true), format, fs, MakeFileFragments(format, fs, schema));
if (!dataset.ok()) {
WIND_ERR_LOG("MakeDataSet fail, detail:{}", dataset.status().ToString());
return nullptr;
}

return dataset.ValueUnsafe();
}

std::shared_ptr MakeScanOptions(const wind::thrift::ReadLineNode& node,
const std::shared_ptr& ds) {
auto options = std::make_shared();
options->use_threads = true;
arrow::io::IOContext io_context(arrow::default_memory_pool(), ::arrow::internal::GetCpuThreadPool());
options->io_context = io_context;

std::vector columns;
for (const auto& column : node.columns) {
columns.push_back(schema_.fields[column].name);
}

// The CsvFileFragment wouldn't benefit from this as much, but
// can still benefit from skipping conversion of unneeded columns.
auto resultProject = ds::ProjectionDescr::FromNames(std::move(columns), *(ds->schema()));
if (!resultProject.ok()) {
// in this case, convert all columns in table
options->projection = cp::project({}, {});
WIND_ERR_LOG("get projectionDescr fail, detail:{}", resultProject.status().ToString());
} else {
auto project = std::move(resultProject).ValueUnsafe();
options->projection = std::move(project.expression);
}

if (!node.__isset.filter) return options;

options->filter = MakeScanFilter(node.filter);
return options;
}
```
The ExecPlan for short of my test is like
```
:ConsumingSinkNode{}
:GroupByNode{keys=["starttime", "cus_area_id", "rat"], aggregates=[
hash_sum(intra_rau_req_times),
....
]}
:ProjectNode{projection=["starttime": get_time(starttime, 86400), cus_area_id, rat...]}
:FilterNode{filter=((get_time(starttime, 86400) >= 1621699200) and (get_time(starttime, 86400) < 1621785600))}
:SourceNode{}
```

In conclusion, my questions are these(Arrow version: 10.0.0):
1. Is that add multi-connections to HDFS will help?
2. Am I wrong with using Arrow API so that I can't get improvement when I increase the IO thread pool num?
3. How can I get the performance improvement of reading CSV from HDFS in this case?

Any hint is welcome! Thx in advance!

### Component(s)

C++

Contributor guide

Open the contributing guide

Research direction

No specific files or tests are named. Start by reproducing the Arrow 10.0.0 C++ dataset scan with ARROW_IO_THREADS set to 1, 2, and 3, then trace the HDFS read interface and connection usage. Done would require a maintainer-approved scope, implementation approach, and benchmark showing whether concurrent HDFS reads improve CSV scan performance.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp
Domain
data-engineering, distributed-systems, performance
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.