Add binning Runnable. (#85)

* Add the binning Runnable - Calculate the basic statistics results and binning stats results for multiple columns.

* Remove the unused methods.

* Remove unnecessary print.

* Add new line.

* Remove empty file.

Author: brightcoder01

runnables/binning.py

@@ -0,0 +1,51 @@
+import argparse
+import mars.dataframe as md
+import os
+from binning.binning import calc_stats
+from run_io.db_adapter import convertDSNToRfc1738
+from sqlalchemy import create_engine
+
+
+def build_argument_parser():
+    parser = argparse.ArgumentParser(allow_abbrev=False)
+    parser.add_argument("--dbname", type=str, required=True)
+    parser.add_argument("--columns", type=str, required=True)
+    parser.add_argument("--bin_methods", type=str, required=False)
+    parser.add_argument("--bin_nums", type=str, required=False)
+    parser.add_argument("--reverse_cumsum", type=bool, default=False)
+
+    return parser
+
+
+if __name__ == "__main__":
+    parser = build_argument_parser()
+    args, _ = parser.parse_known_args()
+    columns = args.columns.split(',')
+    bin_methods = args.bin_methods.split(',') if args.bin_methods else None
+    bin_nums = [int(item) for item in args.bin_nums.split(',')] if args.bin_nums else None
+
+    select_input = os.getenv("SQLFLOW_TO_RUN_SELECT")
+    output = os.getenv("SQLFLOW_TO_RUN_INTO")
+    datasource = os.getenv("SQLFLOW_DATASOURCE")
+
+    url = convertDSNToRfc1738(datasource, args.dbname)
+    engine = create_engine(url)
+    input_md = md.read_sql(
+        sql=select_input,
+        con=engine)
+    input_md.execute()
+
+    stats_df = calc_stats(
+        input_md,
+        columns,
+        bin_methods,
+        bin_nums,
+        {},
+        args.reverse_cumsum)
+
+    print("Persist the statistics result into the table {}".format(output))
+    stats_df.to_sql(
+        name=output,
+        con=engine,
+        index=False
+    )

runnables/binning/__init__.py

@@ -0,0 +1,112 @@
+import mars.dataframe as md
+import mars.tensor as mt
+import numpy as np
+import pandas as pd
+
+
+class BinningMethod(object):
+    BUCKET = "bucket"
+    QUANTILE = "quantile"
+    LOG_BUCKET = "log_bucket"
+
+
+def binning(
+    in_md,
+    col_name,
+    bin_method,
+    bins,
+    boundaries):
+    if boundaries:
+        bin_o, bins = md.cut(in_md[col_name], bins=boundaries, labels=False, retbins=True)
+        bins_np = bins.to_numpy()
+    else:
+        if bin_method.lower() == BinningMethod.BUCKET.lower():
+            bin_o, bins = md.cut(in_md[col_name], bins=bins, labels=False, retbins=True)
+            bins_np = bins.to_numpy()
+        elif bin_method.lower() == BinningMethod.LOG_BUCKET.lower():
+            bin_o, bins = md.cut(mt.log(in_md[col_name]), bins=bins, labels=False, retbins=True)
+            bins_np = np.exp(bins.to_numpy())
+        else:
+            raise ValueError("Unsupport binning method: {}".format(bin_method))
+
+    return bin_o, bins_np
+
+
+def cumsum(arr, reverse):
+    if type(arr) == np.ndarray:
+        sum_arr = arr
+    elif type(arr) == pd.DataFrame:
+        sum_arr = arr.to_numpy()
+    else:
+        raise ValueError("Invalid input type: {}".format(type(arr)))
+
+    for i in range(np.ndim(arr)):
+        sum_arr = np.flip(np.cumsum(np.flip(sum_arr, i), i), i) if reverse else np.cumsum(sum_arr, i)
+
+    if type(arr) == np.ndarray:
+        return sum_arr
+    elif type(arr) == pd.DataFrame:
+        return pd.DataFrame(sum_arr)
+    else:
+        raise ValueError("Invalid input type: {}".format(type(arr)))
+
+
+def calc_binning_stats(
+    in_md,
+    sel_cols,
+    bin_methods,
+    bin_nums,
+    cols_bin_boundaries,
+    reverse_cumsum=False):
+    cols_bin_stats = []
+    for i in range(len(sel_cols)):
+        sel_col = sel_cols[i]
+        bin_o, bins = binning(in_md, sel_col, bin_methods[i], bin_nums[i], cols_bin_boundaries.get(sel_col, None))
+        bin_num = len(bins) - 1
+        bin_prob_df = bin_o.value_counts(normalize=True).to_pandas().to_frame()
+        bin_prob_df = bin_prob_df.reindex(range(bin_num), fill_value=0)
+        bin_cumsum_prob_df = cumsum(bin_prob_df, reverse_cumsum)
+
+        cols_bin_stats.append(
+            {
+                "name": sel_col,
+                "bin_boundaries": ','.join(bins.astype(str)),
+                "bin_prob": ','.join(bin_prob_df[bin_prob_df.columns[0]].to_numpy().astype(str)),
+                "bin_cumsum_prob": ','.join(bin_cumsum_prob_df[bin_cumsum_prob_df.columns[0]].to_numpy().astype(str))
+            }
+        )
+
+    return pd.DataFrame(cols_bin_stats)
+
+
+def calc_basic_stats(
+    in_md,
+    sel_cols):
+    stats_data = [
+        {
+            "name": sel_col,
+            "min": mt.min(in_md[sel_col]).to_numpy(),
+            "max": mt.max(in_md[sel_col]).to_numpy(),
+            "mean": mt.mean(in_md[sel_col]).to_numpy(),
+            "median": mt.median(in_md[sel_col]).to_numpy(),
+            "std": mt.std(in_md[sel_col]).to_numpy(),
+        } for sel_col in sel_cols
+    ]
+
+    return pd.DataFrame(stats_data)
+
+
+def calc_stats(
+    in_md,
+    sel_cols,
+    bin_methods,
+    bin_nums,
+    cols_bin_boundaries,
+    reverse_cumsum=False):
+    basic_stats_df = calc_basic_stats(in_md, sel_cols)
+    cols_bin_stats_df = calc_binning_stats(in_md, sel_cols, bin_methods, bin_nums, cols_bin_boundaries, reverse_cumsum)
+    
+    stats_df = pd.merge(basic_stats_df, cols_bin_stats_df, how='inner', on='name')
+
+    return stats_df
+

runnables/binning/binning.py

@@ -1,3 +1,4 @@
 tsfresh
 sqlalchemy
-mysql
+mysql
+pymars