[jvm-packages] Fix #3489: Spark repartitionForData can potentially sh…

[weitian]

ranker training needs grouped data and the order of labeledPoint within a group has to be kept for evaluation.
For distributed training, repartition train data is important to keep the workload balanced. But the RDD repartition here shuffles all data, which breaks group and loses ordering required for ranking.

Also the random splitting the test data out train data here breaks the group, too.

The fix is to group the train data as RDD[Array[XGBLabeledPoint]] before the repartitioning and splitting.

While loading training data into RDD partitions, we could not control the edge of each partition, and the LabeledPoints belonging to one group may be broken into two RDD partitions. The chunks are stitched together by group id.

Another challenge is to conserve the JVM heap memory usage, so we can leave more memory to the XGBoost native library for training. Based on the streaming feature of Scala, the Iterator of a collection is "view based" rather than "builder based".
It means the Scala Iterator is lazy-evaluation which does not load all data into memory.
So we need to keep all transformation steps lazy (Spark transformation is lazy by default). This is the design idea for LabeledPointGroupIterator.

Based on same reason, duplication of Iterator here is not very memory efficient since the gap between two iterators are cached in memory.

[weitian]

The build was passed the first time, but it failed after I closed / reopen the PR

[CodingCat]

something wrong when downloading file from github....retriggered the test

[CodingCat]

I took the first round of review and have some questions about the correctness of the algorithm

[CodingCat]

the new changes lead to the stuck of some test case?

[weitian]

the new changes lead to the stuck of some test case?

hmm, all test passed in my local build.

the missing test starts from "distributed training with the specified worker number"
And this test does not have grouped data.

Here is my local log:
Tracker started, with env={DMLC_NUM_SERVER=0, DMLC_TRACKER_URI=10.0.2.15, DMLC_TRACKER_PORT=9091, DMLC_NUM_WORKER=3}
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-13075541539431860144/train.row.page
[14:54:14] SparsePageSource: Finished writing to /tmp/1-cache-13075541539431860144/train
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-13075541539431860144/test.row.page
[14:54:14] SparsePageSource: Finished writing to /tmp/1-cache-13075541539431860144/test
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-28775758788172922589/train.row.page
[14:54:14] SparsePageSource: Finished writing to /tmp/1-cache-28775758788172922589/train
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-28775758788172922589/test.row.page
[14:54:14] SparsePageSource: Finished writing to /tmp/1-cache-28775758788172922589/test
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-02848682310356378771/train.row.page
[14:54:14] SparsePageSource: Finished writing to /tmp/1-cache-02848682310356378771/train
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-02848682310356378771/test.row.page
[14:54:14] SparsePageSource: Finished writing to /tmp/1-cache-02848682310356378771/test
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-13075541539431860144/train.col.page
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-02848682310356378771/train.col.page
[14:54:14] SparsePage::Writer Finished writing to /tmp/1-cache-28775758788172922589/train.col.page

• distributed training with the specified worker number

Compared with Javis log:
Tracker started, with env={DMLC_NUM_SERVER=0, DMLC_TRACKER_URI=10.20.0.139, DMLC_TRACKER_PORT=9091, DMLC_NUM_WORKER=3}
[23:10:31] SparsePage::Writer Finished writing to /tmp/1-cache-18689073119010624543/train.row.page
[23:10:31] SparsePageSource: Finished writing to /tmp/1-cache-18689073119010624543/train
[23:10:31] SparsePage::Writer Finished writing to /tmp/1-cache-18689073119010624543/test.row.page
[23:10:31] SparsePageSource: Finished writing to /tmp/1-cache-18689073119010624543/test
[23:10:31] SparsePage::Writer Finished writing to /tmp/1-cache-08996467644075568956/train.row.page
[23:10:31] SparsePageSource: Finished writing to /tmp/1-cache-08996467644075568956/train
[23:10:31] SparsePage::Writer Finished writing to /tmp/1-cache-08996467644075568956/test.row.page
[23:10:31] SparsePageSource: Finished writing to /tmp/1-cache-08996467644075568956/test
No output has been received in the last 10m0s, this potentially indicates a stalled build or something wrong with the build itself.

It looks like the cache writing was not finished. (with "useExternalMemory = true") ?

[weitian]

re-trigger the build

[weitian]

@CodingCat the reason that java test hung was that I set "nWorkers = 3" in the test which exceeded the CPU cores in testing container. I changed it to "nWorkers = numWorkers"

[weitian]

@CodingCat Need some help here. Do you have any idea why the python-test failed?

[hcho3]

@weitian I restarted the test. It had failed due to random out of memory error.

[weitian]

@hcho3 pre-merge still failed, any idea?

[hcho3]

@weitian Tests have been flaky lately. I'll need to take a look.

[hcho3]

Why did you restart the tests? Jenkins was still running

[weitian]

@hcho3 @CodingCat
Any comment?

[hcho3]

@weitian I have reset the Jenkins environment to get rid of residual files that was causing conflicts.

@CodingCat Does it look good to you? Should we merge it once tests are fixed?

[CodingCat]

I haven't got bandwidth to take a second round on this, maybe @yanboliang can help on reviewing

[CodingCat]

I took a quick review of the current approach, the assumptions here still raise some concerns to me

how about we make it as simple as

1. if it's ranking training application, we just do not reshuffle the input rdd but instead starting N workers for the user (N is the number of partitions in the input rdd) if N doesn't match with nWorkers set by the user, we can give a WARN log

OR

2. if it's ranking training application, we just do not reshuffle the input rdd, if user's configuration of nWorkers doesn't match with the partition number of input rdd, we stop the application with the reason giving to the user....

[CodingCat]

is this assumption true? what if we have 4 partitions each with millions of records belonging to a single group?

[weitian]

Let's do some math here.

Assume that we have a 100GB training data, the HDFS block size is 128MB, we have 800 partitions when loading data into RDD.
for each RDD partition, there are two edge groups, the first group and the last group, so the total number of edge groups are 800 * 2, which is small enough to be coalesced to one executor and processed.

What is an edge group? See if a group has 100 LabeledPoint, 30 of them may in one RDD partition, the 70 of them may in the next RDD partition. So there are only two edge groups in each partition.

The millions of groups other then the first and last groups in a RDD partition are normal groups and we do not need to stitch them.

[CodingCat]

how you guarantee

1. the data is loaded as plain HDFS file

2. user didn't apply any additional transformation including repartition before feeding as training data

[weitian]

I just use HDFS as an example, since we only collect two groups from each partition, the total size of edge groups is relatively small. So this is actually a optimization for performance, not causing performance issue.

Or we can use RDD group by which actually does the same logic but in a distributed fashion.

[CodingCat]

you still didn’t answer my previous question, how about the user pass in a RDD containing 4 partitions each of which contains only 1/2 group data,

All groups will be collected into a single partition?

[weitian]

Yes, all data are collection into one partition, and then produces a list of groups in one partition.

[hcho3]

@weitian Please ignore the failed test continuous-integration/jenkins/pr-head. (This was due to misconfiguration of Jenkins CI server.) Only continuous-integration/jenkins/pr-merge is relevant.

[weitian]

@CodingCat To push the group partition logic to the user may not be a good idea.

Sharing my own experience, when I started to use the xgboost-spark package, I just called the XGBoostRegressor API, and there was no error or exception. I found this bug until I saw the metrics which was very different from the result of the model trained in a single node.

Since evenly partitioned data is very critical for parallel training, the RDD has to be repartitioned and user has to deal with the crossing border issue with groups, which is very tricky.

Another issue is that, even the data is pre-partitioned, the current logic to split the training and testing sets still mess up the groups.

I know this is a PR with a lot of changes, but logically all those fixes in the PR are connected and it is hard to break them into multiple PRs.

[CodingCat]

@weitian it's not about that this PR has many changes, it's that some assumptions which this PR is based on either bring potentially-incorrect results (previous dropping group version) or performance bottleneck (current coalesce(1) version)

my suggestion is

1. disable training and test split for ranking task

2. validate if the input RDD follows the constraints for ranking task

2.1) if not follow, by default, stop the application and give the reason

2.2) user can switch to the dropping group version if they know what they are doing

[weitian]

@CodingCat

1. I changed coalesce(1) to use standard RDD groupBy. There should be no performance issue.
2. disable training and test split gives user a bad experience since the metrics on test split is needed.
3. It is very hard to explain to the user that there are so many restrictions if she/he want to training a ranker. And user will be confused on how to prepare the training data.
4. without group, user can not do pairwise.

[hcho3]

I have filed an issue #3720 to address flaky tests once and for all.

[CodingCat]

generally LGTM, more comments though

[CodingCat]

is there any making-sense but small dataset we can use for ranking unit test? I think the current unit test regarding ranking are all testing if the training can finished

[weitian]

This PR is to make sure that, after training data is repartitioned and split, the training groups are still correct and the order of LabeledPoints within a group are kept.

My changed unit test has a good coverage on the corner cases: test("repartitionForTrainingGroup with group data")

Yes, it only tests if training can finished. I can not use my training data from Apple, so let's see if we could find some sample data set for ranking.

[weitian]

@hcho3 still some failure in AppVeyor build. Any idea?

[hcho3]

@weitian That was a weird failure. I think it was due to spurious download failure for R packages.

[CodingCat]

thanks, merged !