code: 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}