def METHOD_NAME(
    tmpdir,
    input_df,
    config,
    balance,
    num_cpus=2,
    num_gpus=None,
):
    with ray_start(num_cpus=num_cpus, num_gpus=num_gpus):
        csv_filename = os.path.join(tmpdir, "dataset.csv")
        input_df.to_csv(csv_filename)
        dataset_parquet = create_data_set_to_use("parquet", csv_filename)
        model = LudwigModel(config, backend=RAY_BACKEND_CONFIG, callbacks=None)
        output_dir = None
        try:
            _, output_dataset, output_dir = model.train(
                dataset=dataset_parquet,
                training_set=None,
                validation_set=None,
                test_set=None,
                skip_save_processed_input=True,
                skip_save_progress=True,
                skip_save_unprocessed_output=True,
                skip_save_log=True,
            )
        finally:
            # Remove results/intermediate data saved to disk
            shutil.rmtree(output_dir, ignore_errors=True)
        input_train_set = input_df.sample(frac=0.7, replace=False)
        processed_len = output_dataset[0].ds.count()
        processed_target_pos = output_dataset[0].ds.sum(on="Label_mZFLky")
        processed_target_neg = output_dataset[0].ds.count() - output_dataset[0].ds.sum(on="Label_mZFLky")
        assert len(input_train_set) == 140
        assert 0.05 <= len(input_train_set[input_train_set["Label"] == 1]) / len(input_train_set) <= 0.15
        assert round(processed_target_pos / processed_target_neg, 1) == 0.5
        assert model.backend.df_engine.parallelism == RAY_BACKEND_CONFIG["processor"]["parallelism"]
        assert isinstance(model.backend, RayBackend)
        if balance == "oversample_minority":
            assert len(input_train_set) < processed_len
        if balance == "undersample_majority":
            assert len(input_train_set) > processed_len