add PySparkOvertimeMonitor to avoid exceeding time budget

[levscaut]

@thinkall

Why are these changes needed?

For very large dataset that takes very long time to complete a trial, the actual running time may be way longer than the time budget. It's usually the case that when the time budget is about to run out, a new trial is still started, and the current manage system will need to wait till the new trial is completed. In the worst case, the actual running time will be time budget plus time for a trial. For large dataset this extra time cost is unacceptable.

To make the time cost meet the time budget requirement, let me introduce PySparkOvertimeMonitor. It's a context manager that creates a monitor thread to terminate Spark jobs in time when the running time has exceeded the time budget. When the running jobs are cancelled, you can still fetch the last finished result.

You can set the argument force_cancel to be True in automl settings to enable the monitor. It's set to False by default.

    automl_experiment = AutoML()
    automl_settings = {
        ...,
        "force_cancel": True,
    }

    automl_experiment.fit(**automl_settings)

I put PySparkOvertimeMonitor in the flaml.tune.spark.utils. Acting as a context manager, I wrapped the running parts of both sequential and spark parallel in flaml.tune with it. Don't worry, it will not function if the force_cancel is set to False or the Spark is not installed.

with PySparkOvertimeMonitor(time_start, time_budget_s, force_cancel, parallel=parallel):
    results = parallel(
            delayed(evaluation_function)(trial_to_run.config)
            for trial_to_run in trials_to_run)

with PySparkOvertimeMonitor(time_start, time_budget_s, force_cancel):
    result = evaluation_function(trial_to_run.config)

I also wrote a test called test_overtime in test/spark/test_overtime.py . In this test file you can see the actual running time only exceeds the time budget by 0.2s.

Checks

• I've used pre-commit to lint the changes in this PR, or I've made sure lint with flake8 output is two 0s.
• [] I've included any doc changes needed for https://microsoft.github.io/FLAML/. See https://microsoft.github.io/FLAML/docs/Contribute#documentation to build and test documentation locally.
• I've added tests (if relevant) corresponding to the changes introduced in this PR.
• I've made sure all auto checks have passed.

[levscaut]

@microsoft-github-policy-service agree company="Microsoft"

[thinkall]

@levscaut Thank you for the contribution!

Could you remove the file automl_pyspark_df.py:Zone.Identifier

https://github.com/microsoft/FLAML/actions/runs/4192036436/jobs/7267230259#step:2:425

[thinkall]

Also, please remove unnecessary changes in test folders, only check in your code/docs changes.

[levscaut]

@thinkall Thanks for the advice! I have committed a cleaner version.

[thinkall]

A test on macos failed: https://github.com/microsoft/FLAML/actions/runs/4219786733/jobs/7325478207#step:6:84
@levscaut could you fix it? Thanks.

[thinkall]

FAILED test/nlp/test_autohf.py::test_hf_data - ZeroDivisionError: float division by zero
FAILED test/spark/test_overtime.py::test_overtime - assert 1.2553856372833252 < 1
 +  where 1.2553856372833252 = abs((11.255385637283325 - 10))

in https://github.com/microsoft/FLAML/actions/runs/4220014982/jobs/7325940952

[qingyun-wu]

Looks good to me.