def fit(dataset, alpha=1e-8, max_iterations=10, save_results=True, show=False):
    from disco.worker.pipeline.worker import Worker, Stage
    from disco.core import Job, result_iterator
    import numpy as np
    if dataset.params["y_map"] == []:
        raise Exception("Logistic regression requires a target label mapping parameter.")
    try:
        alpha = float(alpha)
        max_iterations = int(max_iterations)
        if max_iterations < 1:
            raise Exception("Parameter max_iterations should be greater than 0.")
    except ValueError:
        raise Exception("Parameters should be numerical.")
    thetas = np.zeros(len(dataset.params["X_indices"]) + 1)
    J = [0]
    for i in range(max_iterations):
        job = Job(worker=Worker(save_results=save_results))
        job.pipeline = [
            ("split", Stage("map", input_chain=dataset.params["input_chain"], init=simple_init, process=map_fit)),
            ('group_all', Stage("reduce", init=simple_init, process=reduce_fit, combine=True))]
        job.params = dataset.params
        job.params["thetas"] = thetas
        job.run(name="logreg_fit_iter_%d" % (i + 1), input=dataset.params["data_tag"])
        fitmodel_url = job.wait(show=show)
        for k, v in result_iterator(fitmodel_url):
            if k == "J":
                J.append(v)
            else:
                thetas = v
        if np.abs(J[-2] - J[-1]) < alpha:
            if show:
                print("Converged at iteration %d" % (i + 1))
            break
    return {"logreg_fitmodel": fitmodel_url}