code: def reduce_fit(interface, state, label, inp):
      from disco.util import kvgroup
      import numpy as np
      out = interface.output(0)
      fit_model = {"y_labels": [], "y_sum": 0, "iv": set()}
      combiner = {}
      means, variances = [], []
      k_prev = ""
      for key, value in kvgroup(inp):
          k_split = key.split(state["delimiter"])
          if len(k_split) == 3:
              fit_model["iv"].add(tuple(k_split[1:]))
              out.add(tuple(k_split), sum(value))
          elif len(k_split) == 2:
              if k_split[0] != k_prev and k_prev != "":
                  mean, var = zip(*[combiner[key] for key in sorted(combiner.keys())])
                  means.append(mean)
                  variances.append(var)
              n_a = mean_a = var_a = 0
              for n_b, mean_b, var_b in value:
                  n_ab = n_a + n_b
                  var_a = ((n_a * var_a + n_b * var_b) / float(n_ab)) + (
                      n_a * n_b * ((mean_b - mean_a) / float(n_ab)) ** 2)
                  mean_a = (n_a * mean_a + n_b * mean_b) / float(n_ab)
                  n_a = n_ab
              combiner[int(k_split[1])] = (mean_a, var_a + 1e-9)
              k_prev = k_split[0]
          else:
              fit_model[key] = np.sum(value)
              fit_model["y_sum"] += fit_model[key]
              fit_model["y_labels"].append(key)
      if len(means) > 0:
          mean, var = zip(*[combiner[key] for key in sorted(combiner.keys())])
          out.add("mean", np.array(means + [mean], dtype=np.float32))
          variances = np.array(variances + [var], dtype=np.float32)
          out.add("var", variances)
          out.add("var_log", np.log(np.pi * variances))
      prior = [fit_model[y_label] / float(fit_model["y_sum"]) for y_label in fit_model["y_labels"]]
      out.add("prior", np.array(prior, dtype=np.float32))
      out.add("prior_log", np.log(prior))
      out.add("iv", list(fit_model["iv"]))
      out.add("y_labels", fit_model["y_labels"])