You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@madlib.apache.org by ri...@apache.org on 2015/12/01 22:08:11 UTC
incubator-madlib git commit: SVM: Add CV support with generic class
Repository: incubator-madlib
Updated Branches:
refs/heads/master 98b088dba -> 72966db43
SVM: Add CV support with generic class
JIRA: MADLIB-915
PR: Closes #4
Authors:
Xiaocheng Tang <xi...@gmail.com>
Rahul Iyer <ra...@gmail.com>
Changes:
- Add cross validation support on lambda, epsilon, init_stepsize,
max_iter, and decay_factor
- Add support for optionally writing validation results to a sql table
- Add support for lazy-generation of cv datasets
- Add internal generic CrossValidator class which is used for implementing
this issue
- Refactoring SVM for better modularity
- Ignore cv on epsilon for classification
- Cross validation now works when independent variables are queries
- Fixed "zero length field name in format" error in python < 2.7
Project: http://git-wip-us.apache.org/repos/asf/incubator-madlib/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-madlib/commit/72966db4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-madlib/tree/72966db4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-madlib/diff/72966db4
Branch: refs/heads/master
Commit: 72966db4325d4c21cb4bc394a67df8959d998d79
Parents: 98b088d
Author: Xiaocheng Tang <xt...@pivotal.io>
Authored: Tue Dec 1 13:04:05 2015 -0800
Committer: Rahul Iyer <ri...@pivotal.io>
Committed: Tue Dec 1 13:06:25 2015 -0800
----------------------------------------------------------------------
src/ports/postgres/modules/svm/svm.py_in | 751 +++++++++++--------
.../postgres/modules/svm/test/linear_svm.sql_in | 79 +-
.../utilities/in_mem_group_control.py_in | 8 +-
.../postgres/modules/utilities/utilities.py_in | 11 +
.../modules/validation/cross_validation.py_in | 18 +-
.../postgres/modules/validation/cv_utils.py_in | 1 +
.../modules/validation/internal/__init__.py_in | 0
.../validation/internal/cross_validation.py_in | 420 +++++++++++
8 files changed, 976 insertions(+), 312 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/svm/svm.py_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/svm/svm.py_in b/src/ports/postgres/modules/svm/svm.py_in
index c543a4a..20c6d7f 100644
--- a/src/ports/postgres/modules/svm/svm.py_in
+++ b/src/ports/postgres/modules/svm/svm.py_in
@@ -1,3 +1,5 @@
+from __future__ import division
+
import plpy
from utilities.control import MinWarning
@@ -9,6 +11,7 @@ from utilities.utilities import add_postfix
from utilities.utilities import _string_to_array_with_quotes
from utilities.utilities import _string_to_array
from utilities.utilities import _assert
+from utilities.utilities import num_features
from utilities.validate_args import cols_in_tbl_valid
from utilities.validate_args import input_tbl_valid
@@ -16,6 +19,8 @@ from utilities.validate_args import output_tbl_valid
from utilities.validate_args import is_var_valid
from utilities.validate_args import get_expr_type
+from validation.internal.cross_validation import CrossValidator
+
def _compute_svm(args):
"""
@@ -23,6 +28,8 @@ def _compute_svm(args):
@return Number of iterations that has been run
"""
+ init_stepsize = args['init_stepsize']
+ args['stepsize'] = init_stepsize
iterationCtrl = GroupIterationController(args)
with iterationCtrl as it:
it.iteration = 0
@@ -45,20 +52,203 @@ def _compute_svm(args):
""")
it.info()
if it.kwargs['decay_factor'] > 0:
- it.kwargs['stepsize'] = it.kwargs['stepsize'] * it.kwargs['decay_factor']
+ it.kwargs['stepsize'] *= it.kwargs['decay_factor']
else:
- it.kwargs['stepsize'] = it.kwargs['init_stepsize'] / (it.iteration + 1)
+ it.kwargs['stepsize'] = init_stepsize / (it.iteration + 1)
has_converged = it.test(
- """
- {iteration} >= {max_iter}
- OR {schema_madlib}.internal_linear_svm_igd_distance(
- _state_previous, _state_current) < {tolerance}
- """)
+ """
+ {iteration} >= {max_iter}
+ OR {schema_madlib}.internal_linear_svm_igd_distance(
+ _state_previous, _state_current) < {tolerance}
+ """)
it.final()
return iterationCtrl.iteration
# ---------------------------------------------------
+def _verify_table(source_table, model_table, dependent_varname,
+ independent_varname, **kwargs):
+ # validate input
+ input_tbl_valid(source_table, 'SVM')
+ _assert(is_var_valid(source_table, dependent_varname),
+ "SVM error: invalid dependent_varname "
+ "('{dependent_varname}') for source_table "
+ "({source_table})!".format(dependent_varname=dependent_varname,
+ source_table=source_table))
+ _assert(is_var_valid(source_table, independent_varname),
+ "SVM error: invalid independent_varname "
+ "('{independent_varname}') for source_table "
+ "({source_table})!".format(independent_varname=independent_varname,
+ source_table=source_table))
+
+ dep_type = get_expr_type(dependent_varname, source_table)
+ if '[]' in dep_type:
+ plpy.error("SVM error: dependent_varname cannot be of array type!")
+
+ # validate output tables
+ output_tbl_valid(model_table, 'SVM')
+ summary_table = add_postfix(model_table, "_summary")
+ output_tbl_valid(summary_table, 'SVM')
+
+
+def _verify_grouping(schema_madlib, source_table, grouping_col):
+ if grouping_col:
+ cols_in_tbl_valid(source_table,
+ _string_to_array_with_quotes(grouping_col),
+ 'SVM')
+ intersect = frozenset(
+ _string_to_array(grouping_col)).intersection(
+ frozenset(
+ ('coef', '__random_feature_data',
+ '__random_feature_data', 'loss'
+ 'num_rows_processed', 'num_rows_skipped',
+ 'norm_of_gradient', 'num_iterations')))
+ _assert(len(intersect) == 0,
+ "SVM error: Conflicting grouping column name.\n"
+ "Some predefined keyword(s) ({0}) are not allowed "
+ "for grouping column names!".format(', '.join(intersect)))
+
+ grouping_list = [i + "::text"
+ for i in explicit_bool_to_text(
+ source_table,
+ _string_to_array_with_quotes(grouping_col),
+ schema_madlib)]
+ grouping_str = ','.join(grouping_list)
+ else:
+ grouping_str = "Null"
+ return grouping_str
+
+
+def _verify_kernel(kernel_func):
+ kernel_func = 'linear' if not kernel_func else kernel_func.lower()
+ # Add non-linear kernels below after implementing them.
+ supported_kernels = ['linear']
+ try:
+ # allow user to specify a prefix substring of
+ # supported kernel function names. This works because the supported
+ # kernel functions have unique prefixes.
+ kernel_func = next(x for x in supported_kernels
+ if x.startswith(kernel_func))
+ except StopIteration:
+ # next() returns a StopIteration if no element found
+ plpy.error("SVM Error: Invalid kernel function: "
+ "{0}. Supported kernel functions are ({1})"
+ .format(kernel_func, ','.join(sorted(supported_kernels))))
+ return kernel_func
+
+
+def _verify_params_dict(params_dict):
+ _assert(not hasattr(params_dict['lambda'], '__len__'),
+ "SVM Error: lambda should not be a list after cross validation!")
+ _assert(not hasattr(params_dict['epsilon'], '__len__'),
+ "SVM Error: epsilon should not be a list after cross validation!")
+ _assert(not hasattr(params_dict['init_stepsize'], '__len__'),
+ "SVM Error: init_stepsize should not be a "
+ "list after cross validation!")
+ _assert(not hasattr(params_dict['decay_factor'], '__len__'),
+ "SVM Error: decay_factor should not be a "
+ "list after cross validation!")
+ _assert(not hasattr(params_dict['max_iter'], '__len__'),
+ "SVM Error: max_iter should not be a list after cross validation!")
+ return params_dict
+
+
+def _build_output_tables(n_iters_run, model_table, args):
+ grouping_col = args['grouping_col']
+ dependent_varname = args['col_dep_var']
+ independent_varname = args['col_ind_var']
+ source_table = args['rel_source']
+ col_grp_key = args['col_grp_key']
+ groupby_str, grouping_str1, using_str = "", "", "ON TRUE"
+ if grouping_col:
+ groupby_str = "GROUP BY {grouping_col}, {col_grp_key}".format(
+ grouping_col=grouping_col, col_grp_key=col_grp_key)
+ grouping_str1 = grouping_col + ","
+ using_str = "USING ({col_grp_key})".format(col_grp_key=col_grp_key)
+ # organizing results
+ dep_type = get_expr_type(dependent_varname, source_table)
+ model_table_query = """
+ CREATE TABLE {model_table} AS
+ SELECT
+ {grouping_str1}
+ (result).coefficients AS coef,
+ (result).loss AS loss,
+ (result).norm_of_gradient AS norm_of_gradient,
+ {n_iters_run} AS num_iterations,
+ (result).num_rows_processed AS num_rows_processed,
+ n_tuples_including_nulls - (result).num_rows_processed
+ AS num_rows_skipped,
+ NULL AS __random_feature_data,
+ ARRAY[{mapping}]::{dep_type}[] AS dep_var_mapping
+ FROM
+ (
+ SELECT
+ {schema_madlib}.internal_linear_svm_igd_result(
+ {col_grp_state}
+ ) AS result,
+ {col_grp_key}
+ FROM {rel_state}
+ WHERE {col_grp_iteration} = {n_iters_run}
+ ) rel_state_subq
+ JOIN
+ (
+ SELECT
+ {grouping_str1}
+ count(*) AS n_tuples_including_nulls,
+ array_to_string(ARRAY[{grouping_str}],
+ ','
+ ) AS {col_grp_key}
+ FROM {source_table}
+ {groupby_str}
+ ) n_tuples_including_nulls_subq
+ {using_str}
+ """.format(n_iters_run=n_iters_run,
+ groupby_str=groupby_str,
+ grouping_str1=grouping_str1,
+ using_str=using_str,
+ source_table=source_table,
+ model_table=model_table,
+ dep_type=dep_type, **args)
+ plpy.execute(model_table_query)
+
+ args['lambda_str'] = str(args['lambda'])
+ summary_table = add_postfix(model_table, "_summary")
+ grouping_text = "NULL" if not grouping_col else grouping_col
+ plpy.execute("""
+ CREATE TABLE {summary_table} AS
+ SELECT
+ '{method}'::text AS method,
+ '__MADLIB_VERSION__'::text AS version_number,
+ '{source_table}'::text AS source_table,
+ '{model_table}'::text AS model_table,
+ '{dependent_varname}'::text AS dependent_varname,
+ '{independent_varname}'::text AS independent_varname,
+ 'linear'::text AS kernel_func,
+ NULL::text AS kernel_params,
+ '{grouping_text}'::text AS grouping_col,
+ 'init_stepsize={init_stepsize}, ' ||
+ 'decay_factor={decay_factor}, ' ||
+ 'max_iter={max_iter}, ' ||
+ 'tolerance={tolerance}'::text AS optim_params,
+ 'lambda={lambda_str}, ' ||
+ 'norm={norm}, ' ||
+ 'n_folds={n_folds}'::text AS reg_params,
+ count(*)::integer AS num_all_groups,
+ 0::integer AS num_failed_groups,
+ sum(num_rows_processed)::bigint AS total_rows_processed,
+ sum(num_rows_skipped)::bigint AS total_rows_skipped,
+ '{epsilon}'::double precision AS epsilon,
+ '{eps_table}'::text AS eps_table
+ FROM {model_table};
+ """.format(grouping_text=grouping_text,
+ summary_table=summary_table,
+ source_table=source_table,
+ model_table=model_table,
+ dependent_varname=dependent_varname,
+ independent_varname=independent_varname,
+ **args))
+
+
def svm(schema_madlib, source_table, model_table,
dependent_varname, independent_varname, kernel_func,
kernel_params, grouping_col, params, is_svc,
@@ -69,218 +259,122 @@ def svm(schema_madlib, source_table, model_table,
# verbosing
verbosity_level = "info" if verbose else "error"
with MinWarning(verbosity_level):
- # validate input
- input_tbl_valid(source_table, 'SVM')
- _assert(is_var_valid(source_table, dependent_varname),
- "SVM error: invalid dependent_varname ('" + str(dependent_varname) +
- "') for source_table (" + source_table + ")!")
- _assert(is_var_valid(source_table, independent_varname),
- "SVM error: invalid independent_varname ('" + str(independent_varname) +
- "') for source_table (" + source_table + ")!")
-
- dep_type = get_expr_type(dependent_varname, source_table)
- if '[]' in dep_type:
- plpy.error("SVM error: dependent_varname cannot be of array type!")
-
- # validate output tables
- output_tbl_valid(model_table, 'SVM')
- summary_table = add_postfix(model_table, "_summary")
- output_tbl_valid(summary_table, 'SVM')
-
- # arguments for iterating
- n_features = plpy.execute("SELECT array_upper({0}, 1) AS dim "
- "FROM {1} LIMIT 1".
- format(independent_varname, source_table)
- )[0]['dim']
- if grouping_col:
- grouping_list = [i + "::text"
- for i in explicit_bool_to_text(
- source_table,
- _string_to_array_with_quotes(grouping_col),
- schema_madlib)]
- grouping_str = ','.join(grouping_list)
- else:
- grouping_str = "Null"
- grouping_str1 = "" if not grouping_col else grouping_col + ","
- grouping_str2 = "1 = 1" if not grouping_col else grouping_col
-
- args = {
- 'rel_args': unique_string(desp='rel_args'),
- 'rel_state': unique_string(desp='rel_state'),
- 'col_grp_iteration': unique_string(desp='col_grp_iteration'),
- 'col_grp_state': unique_string(desp='col_grp_state'),
- 'col_grp_key': unique_string(desp='col_grp_key'),
- 'col_n_tuples': unique_string(desp='col_n_tuples'),
- 'state_type': "double precision[]",
- 'rel_source': source_table,
- 'col_ind_var': independent_varname,
- 'col_dep_var': dependent_varname}
- args.update(locals())
- # variables defined above cannot be moved below this line
- # -------------------------------------------------------
-
- # other params
- kernel_func = 'linear' if not kernel_func else kernel_func.lower()
- # Add non-linear kernels below after implementing them.
- supported_kernels = ['linear']
- try:
- # allow user to specify a prefix substring of
- # supported kernel function names. This works because the supported
- # kernel functions have unique prefixes.
- kernel_func = next(x for x in supported_kernels if x.startswith(kernel_func))
- except StopIteration:
- # next() returns a StopIteration if no element found
- plpy.error("SVM Error: Invalid kernel function: {0}. "
- "Supported kernel functions are ({1})"
- .format(kernel_func, ','.join(sorted(supported_kernels))))
-
- if grouping_col:
- cols_in_tbl_valid(source_table, _string_to_array_with_quotes(grouping_col), 'SVM')
- intersect = frozenset(_string_to_array(grouping_col)).intersection(
- frozenset(
- ('coef', '__random_feature_data',
- '__random_feature_data', 'loss'
- 'num_rows_processed', 'num_rows_skipped',
- 'norm_of_gradient', 'num_iterations')))
- if len(intersect) > 0:
- plpy.error("SVM error: Conflicting grouping column name.\n"
- "Some predefined keyword(s) ({0}) are not allowed!".format(
- ', '.join(intersect)))
-
- args.update(_extract_params(schema_madlib, params))
- args.update(_process_epsilon(is_svc, args))
-
- if not is_svc:
- # transform col_dep_var to binary (1 or -1) if classification
- args.update({
- 'col_dep_var_trans': dependent_varname,
- 'mapping': 'NULL',
- 'method': 'SVR'})
- else:
- # dependent variable mapping
- dep_labels=plpy.execute("""
- SELECT {dependent_varname} AS y
- FROM {source_table}
- WHERE ({dependent_varname}) IS NOT NULL
- GROUP BY ({dependent_varname})
- ORDER BY ({dependent_varname})""".format(**locals()))
- dep_var_mapping = ["'" + d['y'] + "'" if isinstance(d['y'], basestring)
- else str(d['y']) for d in dep_labels]
- if len(dep_var_mapping) != 2:
- plpy.error("SVM error: Classification currently only supports binary output")
-
- col_dep_var_trans = (
- """
- CASE WHEN ({col_dep_var}) IS NULL THEN NULL
- WHEN ({col_dep_var}) = {mapped_value_for_negative} THEN -1.0
- ELSE 1.0
- END
- """
- .format(col_dep_var=dependent_varname,
- mapped_value_for_negative=dep_var_mapping[0])
- )
-
- args.update({
- 'mapped_value_for_negative': dep_var_mapping[0],
- 'col_dep_var_trans': col_dep_var_trans,
- 'mapping': dep_var_mapping[0] + "," + dep_var_mapping[1],
- 'method': 'SVC'})
-
- args['stepsize'] = args['init_stepsize']
- args['is_l2'] = True if args['norm'] == 'l2' else False
-
- # place holder for compatibility
- plpy.execute("CREATE TABLE pg_temp.{0} AS SELECT 1".format(args['rel_args']))
- # actual iterative algorithm computation
- n_iters_run = _compute_svm(args)
-
- # organizing results
- groupby_str = "GROUP BY {grouping_col}, {col_grp_key}".format(**args) if grouping_col else ""
- using_str = "USING ({col_grp_key})".format(**args) if grouping_col else "ON TRUE"
- model_table_query = """
- CREATE TABLE {model_table} AS
- SELECT
- {grouping_str1}
- (result).coefficients AS coef,
- (result).loss AS loss,
- (result).norm_of_gradient AS norm_of_gradient,
- {n_iters_run} AS num_iterations,
- (result).num_rows_processed AS num_rows_processed,
- n_tuples_including_nulls - (result).num_rows_processed
- AS num_rows_skipped,
- NULL AS __random_feature_data,
- ARRAY[{mapping}]::{dep_type}[] AS dep_var_mapping
- FROM
- (
- SELECT
- {schema_madlib}.internal_linear_svm_igd_result(
- {col_grp_state}
- ) AS result,
- {col_grp_key}
- FROM {rel_state}
- WHERE {col_grp_iteration} = {n_iters_run}
- ) rel_state_subq
- JOIN
- (
- SELECT
- {grouping_str1}
- count(*) AS n_tuples_including_nulls,
- array_to_string(ARRAY[{grouping_str}],
- ','
- ) AS {col_grp_key}
- FROM {source_table}
- {groupby_str}
- ) n_tuples_including_nulls_subq
- {using_str}
- """.format(n_iters_run=n_iters_run,
- groupby_str=groupby_str,
- using_str=using_str, **args)
- plpy.execute(model_table_query)
-
- if isinstance(args['lambda'], list):
- args['lambda_str'] = '{' + ','.join(str(e) for e in args['lambda']) + '}'
- else:
- args['lambda_str'] = str(args['lambda'])
-
- plpy.execute("""
- CREATE TABLE {summary_table} AS
- SELECT
- '{method}'::text AS method,
- '__MADLIB_VERSION__'::text AS version_number,
- '{source_table}'::text AS source_table,
- '{model_table}'::text AS model_table,
- '{dependent_varname}'::text AS dependent_varname,
- '{independent_varname}'::text AS independent_varname,
- 'linear'::text AS kernel_func,
- NULL::text AS kernel_params,
- '{grouping_text}'::text AS grouping_col,
- 'init_stepsize={init_stepsize}, ' ||
- 'decay_factor={decay_factor}, ' ||
- 'max_iter={max_iter}, ' ||
- 'tolerance={tolerance}'::text AS optim_params,
- 'lambda={lambda_str}, ' ||
- 'norm={norm}, ' ||
- 'n_folds={n_folds}'::text AS reg_params,
- count(*)::integer AS num_all_groups,
- 0::integer AS num_failed_groups,
- sum(num_rows_processed)::bigint AS total_rows_processed,
- sum(num_rows_skipped)::bigint AS total_rows_skipped,
- '{epsilon}'::double precision AS epsilon,
- '{eps_table}'::text AS eps_table
- FROM {model_table};
- """.format(grouping_text="NULL" if not grouping_col else grouping_col,
- **args))
+ _verify_table(source_table,
+ model_table,
+ dependent_varname,
+ independent_varname)
+ args = locals()
+ args['params_dict'] = _extract_params(schema_madlib, params)
+ _cross_validate_svm(args)
+ _svm_parsed_params(**args)
+
+
+def _cross_validate_svm(args):
+ # updating params_dict will also update args['params_dict']
+ params_dict = args['params_dict']
+
+ if params_dict['n_folds'] > 1 and args['grouping_col']:
+ plpy.error('SVM Error: cross validation '
+ 'with grouping is not supported!')
+
+ cv_params = {}
+ if len(params_dict['lambda']) > 1:
+ cv_params['lambda'] = params_dict['lambda']
+ else:
+ params_dict['lambda'] = params_dict['lambda'][0]
+ if len(params_dict['epsilon']) > 1 and not args['is_svc']:
+ cv_params['epsilon'] = params_dict['epsilon']
+ else:
+ params_dict['epsilon'] = params_dict['epsilon'][0]
+ if len(params_dict['init_stepsize']) > 1:
+ cv_params['init_stepsize'] = params_dict['init_stepsize']
+ else:
+ params_dict['init_stepsize'] = params_dict['init_stepsize'][0]
+ if len(params_dict['max_iter']) > 1:
+ cv_params['max_iter'] = params_dict['max_iter']
+ else:
+ params_dict['max_iter'] = params_dict['max_iter'][0]
+ if len(params_dict['decay_factor']) > 1:
+ cv_params['decay_factor'] = params_dict['decay_factor']
+ else:
+ params_dict['decay_factor'] = params_dict['decay_factor'][0]
+
+ if not cv_params and params_dict['n_folds'] <= 1:
+ # no cross validation
+ return
+
+ if cv_params and params_dict['n_folds'] <= 1:
+ plpy.error("SVM Error: All parameters must be scalar "
+ "or of length 1 when n_folds is 0 or 1")
+
+ if not cv_params and params_dict['n_folds'] > 1:
+ plpy.warning('SVM Warning: n_folds > 1 but no cross validate params provided'
+ 'Ignoring cross validation request.')
+ return
+
+ scorer = 'classification' if args['is_svc'] else 'regression'
+ sub_args = {'params_dict': cv_params}
+ cv = CrossValidator(_svm_parsed_params, svm_predict, scorer, args)
+ val_res = cv.validate(sub_args, params_dict['n_folds']).sorted()
+ val_res.output_tbl(params_dict['validation_result'])
+ params_dict.update(val_res.first('sub_args')['params_dict'])
# ------------------------------------------------------------------------------
+def _svm_parsed_params(schema_madlib, source_table, model_table,
+ dependent_varname, independent_varname, kernel_func,
+ kernel_params, grouping_col, params_dict, is_svc,
+ verbose, **kwargs):
+ """
+ Executes the linear support vector algorithm.
+ """
+ grouping_str = _verify_grouping(schema_madlib, source_table, grouping_col)
+
+ kernel_func = _verify_kernel(kernel_func)
+
+ # arguments for iterating
+ n_features = num_features(source_table, independent_varname)
+
+ args = {
+ 'rel_args': unique_string(desp='rel_args'),
+ 'rel_state': unique_string(desp='rel_state'),
+ 'col_grp_iteration': unique_string(desp='col_grp_iteration'),
+ 'col_grp_state': unique_string(desp='col_grp_state'),
+ 'col_grp_key': unique_string(desp='col_grp_key'),
+ 'col_n_tuples': unique_string(desp='col_n_tuples'),
+ 'state_type': "double precision[]",
+ 'n_features': n_features,
+ 'verbose': verbose,
+ 'is_svc': is_svc,
+ 'schema_madlib': schema_madlib,
+ 'grouping_str': grouping_str,
+ 'grouping_col': grouping_col,
+ 'rel_source': source_table,
+ 'col_ind_var': independent_varname,
+ 'col_dep_var': dependent_varname}
+
+ args.update(_verify_params_dict(params_dict))
+ args.update(_process_epsilon(is_svc, args))
+ args.update(_svc_or_svr(is_svc, source_table, dependent_varname))
+
+ # place holder for compatibility
+ plpy.execute("CREATE TABLE pg_temp.{0} AS SELECT 1"
+ .format(args['rel_args']))
+ # actual iterative algorithm computation
+ n_iters_run = _compute_svm(args)
+ _build_output_tables(n_iters_run, model_table, args)
+
+
def svm_predict(schema_madlib, model_table, new_data_table, id_col_name,
output_table, **kwargs):
- """
- Scores the data points stored in a table using a learned support vector model.
+ """ Scores the data points stored in a table using a
+ learned support vector model.
@param model_table Name of learned model
- @param new_data_table Name of table/view containing the data points to be scored
- @param id_col_name Name of column in source_table containing (integer) identifier for data point
+ @param new_data_table Name of table/view containing the data
+ points to be scored
+ @param id_col_name Name of column in source_table containing
+ (integer) identifier for data point
@param output_table Name of table to store the results
"""
# suppress warnings
@@ -292,8 +386,9 @@ def svm_predict(schema_madlib, model_table, new_data_table, id_col_name,
summary_table = add_postfix(model_table, "_summary")
input_tbl_valid(summary_table, 'SVM')
cols_in_tbl_valid(summary_table,
- ['dependent_varname', 'independent_varname', 'kernel_func',
- 'kernel_params', 'grouping_col'], 'SVM')
+ ['dependent_varname', 'independent_varname',
+ 'kernel_func', 'kernel_params', 'grouping_col'],
+ 'SVM')
# read necessary info from summary
summary = plpy.execute("""
@@ -315,21 +410,22 @@ def svm_predict(schema_madlib, model_table, new_data_table, id_col_name,
input_tbl_valid(new_data_table, 'SVM')
_assert(is_var_valid(new_data_table, dependent_varname),
- "SVM error: dependent_varname ('" + dependent_varname +
+ "SVM Error: dependent_varname ('" + dependent_varname +
"') is invalid for new_data_table (" + new_data_table + ")!")
_assert(is_var_valid(new_data_table, independent_varname),
- "SVM error: independent_varname ('" + independent_varname +
+ "SVM Error: independent_varname ('" + independent_varname +
"') is invalid for new_data_table (" + new_data_table + ")!")
- _assert(id_col_name is not None, "SVM error: id_col_name is NULL!")
+ _assert(id_col_name is not None, "SVM Error: id_col_name is NULL!")
_assert(is_var_valid(new_data_table, id_col_name),
- "SVM error: id_col_name ('" + id_col_name +
+ "SVM Error: id_col_name ('" + id_col_name +
"') is invalid for new_data_table (" + new_data_table + ")!")
output_tbl_valid(output_table, 'SVM')
if method.upper() == 'SVC':
pred_query = """
CASE WHEN {schema_madlib}.array_dot(
- coef::double precision [],
- {independent_varname}::double precision []) >= 0
+ coef::double precision [],
+ {independent_varname}::double precision []
+ ) >= 0
THEN dep_var_mapping[2]
ELSE dep_var_mapping[1]
END
@@ -343,31 +439,27 @@ def svm_predict(schema_madlib, model_table, new_data_table, id_col_name,
""".format(schema_madlib=schema_madlib,
independent_varname=independent_varname)
else:
- plpy.error("SVM error: Invalid 'method' value in summary table. "
+ plpy.error("SVM Error: Invalid 'method' value in summary table. "
"'method' can only be SVC or SVR!")
if grouping_col != "NULL":
sql = """
CREATE TABLE {output_table} AS
SELECT
- {id_col_name} AS id,
+ {id_col_name} AS {id_col_name},
{pred_query} AS prediction,
{model_table}.{grouping_col} as grouping_col
- FROM
- {model_table}
- JOIN
- {new_data_table}
- ON {model_table}.{grouping_col} = {new_data_table}.{grouping_col}
- WHERE
- not {schema_madlib}.array_contains_null({independent_varname})
- ORDER BY
- grouping_col, id
+ FROM {model_table}
+ JOIN {new_data_table}
+ ON {model_table}.{grouping_col} = {new_data_table}.{grouping_col}
+ WHERE not {schema_madlib}.array_contains_null({independent_varname})
+ ORDER BY grouping_col, {id_col_name}
""".format(**locals())
else:
- sql="""
+ sql = """
CREATE TABLE {output_table} AS
SELECT
- {id_col_name} AS id,
+ {id_col_name} AS {id_col_name},
{pred_query} as prediction
FROM
{model_table},
@@ -378,58 +470,115 @@ def svm_predict(schema_madlib, model_table, new_data_table, id_col_name,
plpy.execute(sql)
+def _svc_or_svr(is_svc, source_table, dependent_varname):
+ # transform col_dep_var to binary (1`or -1) if classification
+ _args = {'col_dep_var_trans': dependent_varname,
+ 'mapping': 'NULL',
+ 'method': 'SVR'}
+
+ if is_svc:
+ # dependent variable mapping
+ dep_labels = plpy.execute("""
+ SELECT {dependent_varname} AS y
+ FROM {source_table}
+ WHERE ({dependent_varname}) IS NOT NULL
+ GROUP BY ({dependent_varname})
+ ORDER BY ({dependent_varname})
+ """.format(source_table=source_table,
+ dependent_varname=dependent_varname))
+
+ dep_var_mapping = ["'{0}'".format(d['y'])
+ if isinstance(d['y'], basestring)
+ else str(d['y']) for d in dep_labels]
+
+ _assert(len(dep_var_mapping) == 2,
+ "SVM Error: Classification currently "
+ "only supports binary output!")
+
+ col_dep_var_trans = (
+ """
+ CASE WHEN ({col_dep_var}) IS NULL THEN NULL
+ WHEN ({col_dep_var}) = {mapped_value_for_negative} THEN -1.0
+ ELSE 1.0
+ END
+ """
+ .format(col_dep_var=dependent_varname,
+ mapped_value_for_negative=dep_var_mapping[0])
+ )
+
+ _args.update({
+ 'mapped_value_for_negative': dep_var_mapping[0],
+ 'col_dep_var_trans': col_dep_var_trans,
+ 'mapping': dep_var_mapping[0] + "," + dep_var_mapping[1],
+ 'method': 'SVC'})
+
+ return _args
+
+
def _process_epsilon(is_svc, args):
eps_table = args['eps_table']
grouping_col = args['grouping_col']
grouping_str = args['grouping_str']
col_grp_key = args['col_grp_key']
rel_source = args['rel_source']
+ epsilon = args['epsilon']
rel_epsilon = ''
- select_epsilon = ''
+ select_epsilon = '{0}'.format(epsilon)
as_rel_source = '_src'
- epsilon = args['epsilon']
- if is_svc or not grouping_col or not eps_table:
- if eps_table:
- plpy.warning('SVM: ignoring the input epsilon table!')
- select_epsilon = str(epsilon)
- else:
+ if not is_svc and grouping_col and eps_table:
rel_epsilon = unique_string(desp='rel_epsilon')
input_tbl_valid(eps_table, 'SVM')
_assert(is_var_valid(eps_table, grouping_col),
- "SVM error: invalid column names ('" + str(grouping_col) +
- "') for eps_table (" + eps_table + ")!")
+ "SVM Error: invalid column names ('{grouping_col}') "
+ "for eps_table ('{eps_table}')!"
+ .format(grouping_col=grouping_col,
+ eps_table=eps_table))
plpy.execute("""
DROP TABLE IF EXISTS {rel_epsilon};
CREATE TEMPORARY TABLE {rel_epsilon} AS (
SELECT
{col_grp_key},
coalesce(epsilon, {epsilon}) AS epsilon
- FROM
- (
- SELECT array_to_string(ARRAY[{grouping_str}], ',') AS
+ FROM (
+ SELECT
+ array_to_string(ARRAY[{grouping_str}], ',') AS
{col_grp_key}
- FROM {rel_source}
+ FROM
+ {rel_source}
GROUP BY {grouping_col}
) q1
LEFT JOIN
(
- SELECT array_to_string(ARRAY[{grouping_str}], ',') AS
+ SELECT
+ array_to_string(ARRAY[{grouping_str}], ',') AS
{col_grp_key},
epsilon
FROM {eps_table}
) q2
USING ({col_grp_key})
);
- """.format(**locals()))
-
- select_epsilon = """SELECT epsilon
- FROM
- {rel_epsilon}
- WHERE
- {rel_epsilon}.{col_grp_key} = {as_rel_source}.{col_grp_key}
- """.format(rel_epsilon=rel_epsilon, as_rel_source=as_rel_source,
- col_grp_key=col_grp_key)
+ """.format(rel_epsilon=rel_epsilon,
+ col_grp_key=col_grp_key,
+ epsilon=epsilon,
+ grouping_str=grouping_str,
+ rel_source=rel_source,
+ grouping_col=grouping_col,
+ eps_table=eps_table))
+
+ select_epsilon = (
+ """
+ (
+ SELECT epsilon
+ FROM
+ {rel_epsilon}
+ WHERE
+ {rel_epsilon}.{col_grp_key} = {as_rel_source}.{col_grp_key}
+ )
+ """
+ .format(rel_epsilon=rel_epsilon,
+ as_rel_source=as_rel_source,
+ col_grp_key=col_grp_key))
return {'select_epsilon': select_epsilon,
'epsilon': epsilon,
@@ -440,55 +589,63 @@ def _process_epsilon(is_svc, args):
def _extract_params(schema_madlib, params, module='SVM'):
# NOTICE: the type of values in params_default should be consistent with
# the types specified in params_types
- params_default = {'init_stepsize': 0.01,
- 'decay_factor': 0.9,
- 'max_iter': 100,
- 'tolerance': 1e-10,
- 'lambda': 1.0,
- 'norm': 'L2',
- 'n_folds': 0,
- 'epsilon': 0.01,
- 'eps_table': ''}
-
- params_types = {'init_stepsize': float,
- 'decay_factor': float,
- 'max_iter': int,
- 'tolerance': float,
- 'lambda': list,
- 'norm': str,
- 'n_folds': int,
- 'epsilon': float,
- 'eps_table': str}
+ params_default = {
+ 'init_stepsize': [0.01],
+ 'decay_factor': [0.9],
+ 'max_iter': [100],
+ 'tolerance': 1e-10,
+ 'lambda': [0.01],
+ 'norm': 'L2',
+ 'n_folds': 0,
+ 'validation_result': '',
+ 'epsilon': [0.01],
+ 'eps_table': ''}
+
+ params_types = {
+ 'init_stepsize': list,
+ 'decay_factor': list,
+ 'max_iter': list,
+ 'tolerance': float,
+ 'lambda': list,
+ 'norm': str,
+ 'n_folds': int,
+ 'validation_result': str,
+ 'epsilon': list,
+ 'eps_table': str}
params_vals = extract_keyvalue_params(params,
params_types,
params_default)
-
if params_vals['n_folds'] < 0:
- plpy.error("SVM error: n_folds must be non-negative")
- # FIXME
- _assert(params_vals['n_folds'] <= 1,
- "SVM error: cross-validation has not been implemented")
+ plpy.error("{0} Error: n_folds must be non-negative!".format(module))
# validate lambda
- if hasattr(params_vals['lambda'], '__len__'):
- # this check should be removed after cross validation is added
- if len(params_vals['lambda']) != 1:
- plpy.error("{0} error: lambda must be a scalar or of length 1 when n_folds is 0 or 1".format(module))
- params_vals['lambda'] = params_vals['lambda'][0]
- _assert(params_vals['lambda'] >= 0,
- "SVM error: lambda must be non-negative!")
+ params_vals['lambda'] = map(float, params_vals['lambda'])
+ _assert(all(lmd >= 0 for lmd in params_vals['lambda']),
+ "{0} Error: lambda must be non-negative!".format(module))
+ # validate epsilon
+ params_vals['epsilon'] = map(float, params_vals['epsilon'])
+ _assert(all(e >= 0 for e in params_vals['epsilon']),
+ "{0} Error: epsilon must be non-negative!".format(module))
+ # validating cross validation is delegated to _cross_validate_svm()
+ params_vals['init_stepsize'] = map(float, params_vals['init_stepsize'])
+ _assert(all(e > 0 for e in params_vals['init_stepsize']),
+ "{0} Error: init_stepsize must be positive!".format(module))
+ params_vals['max_iter'] = map(int, params_vals['max_iter'])
+ _assert(all(e > 0 for e in params_vals['max_iter']),
+ "{0} Error: max_iter must be positive!".format(module))
+ params_vals['decay_factor'] = map(float, params_vals['decay_factor'])
+ _assert(all(e <= 1 for e in params_vals['decay_factor']),
+ "{0} Error: decay_factor must be <= 1!".format(module))
+
+ if params_vals['validation_result']:
+ output_tbl_valid(params_vals['validation_result'], 'SVM')
+
params_vals['norm'] = params_vals['norm'].lower()
_assert(params_vals['norm'] == 'l1' or params_vals['norm'] == 'l2',
- "SVM error: norm must be either L1 or L2")
- _assert(params_vals['init_stepsize'] > 0,
- "SVM error: init_stepsize must be positive")
- _assert(params_vals['decay_factor'] <= 1,
- "SVM error: decay_factor must be <= 1")
- _assert(params_vals['max_iter'] > 0,
- "SVM error: max_iter must be positive")
+ "{0} Error: norm must be either L1 or L2!".format(module))
_assert(params_vals['tolerance'] >= 0,
- "SVM error: tolerance must be non-negative")
- _assert(params_vals['epsilon'] >= 0,
- "SVM error: epsilon cannot be less than 0")
+ "{0} error: tolerance must be non-negative!".format(module))
+
+ params_vals['is_l2'] = True if params_vals['norm'] == 'l2' else False
return params_vals
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/svm/test/linear_svm.sql_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/svm/test/linear_svm.sql_in b/src/ports/postgres/modules/svm/test/linear_svm.sql_in
index 0253c77..bba6567 100644
--- a/src/ports/postgres/modules/svm/test/linear_svm.sql_in
+++ b/src/ports/postgres/modules/svm/test/linear_svm.sql_in
@@ -108,7 +108,7 @@ SELECT svm_regression(
NULL,
NULL,
NULL,
- 'init_stepsize=0.01, max_iter=50, lambda=2, norm=l1, epsilon=0.01',
+ 'init_stepsize=0.01, max_iter=50, lambda=2, norm=l2, epsilon=0.01',
false);
DROP TABLE IF EXISTS svr_test_result;
SELECT svm_predict('svr_model', 'svr_train_data', 'id', 'svr_test_result');
@@ -631,3 +631,80 @@ FROM (
where sex = 'M'
) AS q1;
+DROP TABLE IF EXISTS m1, m1_summary;
+SELECT svm_regression(
+ 'svr_train_data',
+ 'm1',
+ 'label',
+ 'ind',
+ NULL,NULL,NULL,
+ 'init_stepsize=0.01, max_iter=3, lambda=[0.0002, 0.2], '
+ 'n_folds=3, epsilon = [0.003, 0.2]',
+ true);
+
+DROP TABLE IF EXISTS m1, m1_summary;
+SELECT svm_regression(
+ 'svr_train_data',
+ 'm1',
+ 'label',
+ 'ind',
+ NULL,NULL,NULL,
+ 'init_stepsize=0.01, max_iter=2, lambda=[0.0002, 0.2], n_folds=3',
+ false);
+-- check which lambda is selected
+SELECT reg_params FROM m1_summary;
+
+-- epsilon values are ignored
+-- the validation table only contains
+-- init_stepsize and lambda
+DROP TABLE IF EXISTS m1, m1_summary, val_res;
+SELECT svm_classification(
+ 'svm_train_data',
+ 'm1',
+ 'label',
+ 'ind',
+ NULL,NULL,NULL,
+ 'init_stepsize=[0.01, 1], max_iter=3, lambda=[20, 0.0002, 0.02], '
+ 'n_folds=3, epsilon=[0.1, 1], validation_result=val_res');
+SELECT * FROM val_res;
+
+DROP TABLE IF EXISTS m1, m1_summary, val_res;
+SELECT svm_classification(
+ 'svm_train_data',
+ 'm1',
+ 'label',
+ 'ind',
+ NULL,NULL,NULL,
+ 'init_stepsize=0.01, max_iter=20, lambda=[20, 0.0002, 0.02], '
+ 'n_folds=3, validation_result=val_res');
+SELECT * FROM val_res;
+-- check which lambda is selected
+SELECT reg_params FROM m1_summary;
+DROP TABLE IF EXISTS svm_test_predict CASCADE;
+SELECT svm_predict('m1','svm_test_data', 'id', 'svm_test_predict');
+-- accuracy with cv
+SELECT
+ count(*) AS misclassification_count
+FROM svm_test_predict NATURAL JOIN svm_test_data
+WHERE prediction <> label;
+
+DROP TABLE IF EXISTS m1, m1_summary;
+SELECT svm_classification(
+ 'svm_train_data',
+ 'm1',
+ 'label',
+ 'ind',
+ NULL,NULL,NULL,
+ 'init_stepsize=0.01, max_iter=20, lambda=0.000002');
+DROP TABLE IF EXISTS svm_test_predict CASCADE;
+SELECT svm_predict('m1','svm_test_data', 'id', 'svm_test_predict');
+-- accuracy without cv
+SELECT
+ count(*) AS misclassification_count
+FROM svm_test_predict NATURAL JOIN svm_test_data
+WHERE prediction <> label;
+
+
+
+
+
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/utilities/in_mem_group_control.py_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/utilities/in_mem_group_control.py_in b/src/ports/postgres/modules/utilities/in_mem_group_control.py_in
index d89212b..08dd269 100644
--- a/src/ports/postgres/modules/utilities/in_mem_group_control.py_in
+++ b/src/ports/postgres/modules/utilities/in_mem_group_control.py_in
@@ -236,7 +236,7 @@ class GroupIterationController:
grouped_state_type=_grouped_state_type)
def __enter__(self):
- verbosity_level = self.kwargs.get('verbosity_level', 'warning')
+ verbosity_level = "info" if self.kwargs['verbose'] else "error"
with MinWarning(verbosity_level):
############################
# create state table
@@ -443,7 +443,7 @@ class GroupIterationController:
self.iteration = self.iteration + 1
group_param = self.group_param
- update_sql = """
+ run_sql = """
SELECT
{_grp_key} AS {col_grp_key},
{grouping_col},
@@ -468,9 +468,7 @@ class GroupIterationController:
select_rel_state=group_param.select_rel_state,
select_n_tuples=group_param.select_n_tuples,
**self.kwargs)
- plpy.info(update_sql)
- update_plan = plpy.prepare(
- update_sql,
+ update_plan = plpy.prepare(run_sql,
["text[]", group_param.grouped_state_type, "text[]", "integer[]"])
res_tuples = plpy.execute(update_plan, [self.new_states.keys,
self.new_states.values,
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/utilities/utilities.py_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/utilities/utilities.py_in b/src/ports/postgres/modules/utilities/utilities.py_in
index e73be2d..31ba6da 100644
--- a/src/ports/postgres/modules/utilities/utilities.py_in
+++ b/src/ports/postgres/modules/utilities/utilities.py_in
@@ -43,6 +43,17 @@ def get_seg_number():
!>)
# ------------------------------------------------------------
+def num_features(source_table, independent_varname):
+ return plpy.execute("SELECT array_upper({0}, 1) AS dim "
+ "FROM {1} LIMIT 1"
+ .format(independent_varname,
+ source_table))[0]['dim']
+# ------------------------------------------------------------
+
+def num_samples(source_table):
+ return plpy.execute("SELECT count(*) AS n FROM {0}"
+ .format(source_table))[0]['n']
+# ------------------------------------------------------------
def unique_string(desp='', **kwargs):
"""
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/validation/cross_validation.py_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/validation/cross_validation.py_in b/src/ports/postgres/modules/validation/cross_validation.py_in
index fce391d..7b39c90 100644
--- a/src/ports/postgres/modules/validation/cross_validation.py_in
+++ b/src/ports/postgres/modules/validation/cross_validation.py_in
@@ -317,15 +317,15 @@ def cross_validation_general(
explore_type_str = "" if not explore_type else "::" + str(explore_type)
# all temporary names
- tbl_all_data = unique_string()
- tbl_train = unique_string()
- tbl_valid = unique_string()
- col_random_id = unique_string()
- tbl_random_id = unique_string()
- tbl_output_model = "pg_temp." + unique_string()
- tbl_output_pred = "pg_temp." + unique_string()
- tbl_output_error = "pg_temp." + unique_string()
- tbl_accum_error = unique_string()
+ tbl_all_data = unique_string(desp='tbl_all_data')
+ tbl_train = unique_string(desp='tbl_train')
+ tbl_valid = unique_string(desp='tbl_valid')
+ col_random_id = unique_string(desp='col_random_id')
+ tbl_random_id = unique_string(desp='tbl_random_id')
+ tbl_output_model = "pg_temp." + unique_string(desp='output_model')
+ tbl_output_pred = "pg_temp." + unique_string(desp='output_pred')
+ tbl_output_error = "pg_temp." + unique_string(desp='output_error')
+ tbl_accum_error = unique_string(desp='accum_error')
tbl_used, col_random_id = _create_data_tbl_id(**locals())
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/validation/cv_utils.py_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/validation/cv_utils.py_in b/src/ports/postgres/modules/validation/cv_utils.py_in
index afd71e8..b6eb601 100644
--- a/src/ports/postgres/modules/validation/cv_utils.py_in
+++ b/src/ports/postgres/modules/validation/cv_utils.py_in
@@ -45,6 +45,7 @@ def __cv_copy_data_with_id(rel_origin, col_data, rel_copied, random_id):
def __cv_copy_data_with_id_compute(rel_origin, col_string, rel_copied, random_id):
plpy.execute("""
+ select setseed(0.5);
drop table if exists {rel_copied};
create temp table {rel_copied} as
select
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/validation/internal/__init__.py_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/validation/internal/__init__.py_in b/src/ports/postgres/modules/validation/internal/__init__.py_in
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/incubator-madlib/blob/72966db4/src/ports/postgres/modules/validation/internal/cross_validation.py_in
----------------------------------------------------------------------
diff --git a/src/ports/postgres/modules/validation/internal/cross_validation.py_in b/src/ports/postgres/modules/validation/internal/cross_validation.py_in
new file mode 100644
index 0000000..2dc6c5f
--- /dev/null
+++ b/src/ports/postgres/modules/validation/internal/cross_validation.py_in
@@ -0,0 +1,420 @@
+import plpy
+from validation.cv_utils import __cv_produce_col_name_string as _cv_col_string
+from validation.cv_utils import __cv_validation_rows as _cv_validation_rows
+from utilities.utilities import __mad_version
+from utilities.utilities import unique_string
+from utilities.utilities import num_samples
+
+from math import sqrt
+from collections import namedtuple
+from functools import partial
+from operator import itemgetter
+from operator import attrgetter
+from itertools import product, repeat, imap, chain
+
+version_wrapper = __mad_version()
+mad_vec = version_wrapper.select_vecfunc()
+
+
+def _extract_data_cols(y, x):
+ """ Extract independent data columns from ARRAY[...] and combine it with dependent data column
+
+ Parameters
+ ==========
+ y : string
+ The dependent data column
+
+ x : string
+ A string of such form: ARRAY[1,?...] where ? indicate 0 or 1 repetitions of the preceding sequence
+
+ Returns
+ =======
+ columns : a list of strings corresponding to column names.
+ """
+ if not x.startswith('ARRAY'):
+ return [y] + x.split(',')
+ import re
+ m = re.search(r'\[((1,)?(?P<cols>.+))\]', x)
+ if not m:
+ plpy.error("SVM error: invalid ({0}) "
+ "for cross validation!".format(x))
+ return [y] + m.group('cols').split(',')
+
+
+class ValidationResult(object):
+ """Wrapper class for cross validation results
+
+ Parameters
+ ==========
+ cv_history : list, optional
+ List of dictionaries.
+ Each dictionary contains the following three keys:
+
+ - mean: float, average of scores using sub_args
+ - std: float, standard deviation of scores using sub_args
+ - sub_args: dict, the values of arguments being validated
+ """
+ def __init__(self, cv_history=None):
+ if cv_history is None:
+ cv_history = []
+ self._cv_history = cv_history
+
+ def _get_leafs(self, sub_args):
+ def _run(sub_args):
+ a = []
+ for k, v in sub_args.iteritems():
+ if isinstance(v, dict):
+ a.extend(_run(v))
+ else:
+ a.append((k, v))
+ return a
+ return _run(sub_args)
+
+ def _flatten(self):
+ a = []
+ for h in self._cv_history:
+ h = dict(h)
+ sub_args = h.pop('sub_args')
+ h.update(dict(self._get_leafs(sub_args)))
+ a.append(h)
+ return a
+
+ def add_one(self, mean, std, sub_args):
+ """Add one record to the history"""
+ record = dict(mean=mean, std=std, sub_args=sub_args)
+ self._cv_history.append(record)
+
+ def sorted(self):
+ """Sort the history w.r.t. mean value and return a new ValidationResult object"""
+ ch = sorted(self._cv_history, reverse=True, key=itemgetter('mean'))
+ return ValidationResult(ch)
+
+ def first(self, attr=None):
+ """Return the attribute of the first record of history
+
+ Parameters
+ ==========
+ attr : string, optional
+ Any string in {'mean', 'std', 'sub_args'} or None
+
+ Returns
+ =======
+ record : dict or float.
+ The return value is either the first record, or the value of the corresponding attr in the first record.
+ """
+ if attr is None:
+ return self._cv_history[0]
+ else:
+ return self._cv_history[0].get(attr)
+
+ def top(self, attr=None):
+ """Return the first after sort"""
+ svr = self.sorted()
+ return svr.first(attr)
+
+ # output cv results as a SQL table
+ def output_tbl(self, tbl_name):
+ """Create a table tbl_name that contains the history
+
+ The columns of tbl_name are mean, std and the leaf keys in sub_args.
+ All column types are assumed to be double precision.
+ """
+ if tbl_name == '' or tbl_name is None:
+ return
+
+ cv_history_f = self._flatten()
+ header = cv_history_f[0].keys()
+ # assuming all keys are string
+ header_str = ','.join(header)
+ # assuming all values are double precision
+ header_with_type_str = ','.join([c + ' double precision'
+ for c in header])
+ plpy.execute("""
+ DROP TABLE IF EXISTS {tbl_name};
+ CREATE TABLE {tbl_name} ({header})
+ """.format(tbl_name=tbl_name,
+ header=header_with_type_str))
+
+ data = []
+ for h in cv_history_f:
+ values = ','.join([str(h[k]) for k in header])
+ data.append("({0})".format(values))
+ data = ','.join(data)
+
+ plpy.execute("""
+ INSERT INTO {tbl_name}({header}) VALUES
+ {data}""".format(data=data,
+ header=header_str,
+ tbl_name=tbl_name))
+
+
+class _ValidationArgs(object):
+ """docstring for _ValidationArgs"""
+ def __init__(self, args):
+ self._args = args
+
+ @classmethod
+ def grid(cls, sub_args):
+ def comb_dict(dicts):
+ return dict(chain.from_iterable(d.iteritems() for d in dicts))
+ def make_dicts(k, vs):
+ return [dict([t]) for t in zip(repeat(k), vs)]
+
+ a = []
+ for k, v in sub_args.iteritems():
+ if isinstance(v, list):
+ a.append(make_dicts(k, v))
+ elif isinstance(v, dict):
+ a.append(make_dicts(k, cls.grid(v)))
+ tuples = product(*a)
+ return map(comb_dict, tuples)
+
+ def make_from(self, **kwargs):
+ def _update_dict(d1, d2):
+ if not isinstance(d1, dict):
+ raise TypeError("{0} is not dict".format(type(d1)))
+ if not isinstance(d2, dict):
+ raise TypeError("{0} is not dict".format(type(d2)))
+ for k, v in d2.iteritems():
+ if isinstance(v, dict) and isinstance(d1[k], dict):
+ _update_dict(d1[k], v)
+ else:
+ d1[k] = v
+ args = dict(self._args)
+ _update_dict(args, kwargs)
+ return args
+
+
+def _cv_copy_data(rel_origin, dependent_varname,
+ independent_varname, rel_copied, random_id):
+ """
+ """
+ target_col, features_col = 'y', 'x'
+ plpy.execute("""
+ select setseed(0.5);
+ drop table if exists {rel_copied};
+ create temp table {rel_copied} as
+ select
+ row_number() over (order by random()) as {random_id},
+ {dependent_varname} as {target_col},
+ {independent_varname} as {features_col}
+ from {rel_origin}
+ """.format(rel_origin=rel_origin,
+ rel_copied=rel_copied,
+ random_id=random_id,
+ dependent_varname=dependent_varname,
+ independent_varname=independent_varname,
+ target_col=target_col, features_col=features_col))
+ return target_col, features_col
+## ========================================================================
+
+
+def _cv_split_data(rel_source, col_data, col_id, row_num,
+ rel_train, rel_valid, n_folds, which_fold):
+ """
+ """
+ col_string = _cv_col_string(rel_source, col_data, [col_id])
+
+ (start_row, end_row) = _cv_validation_rows(row_num, n_folds, which_fold)
+ kwargs = dict(rel_train=rel_train, rel_source=rel_source,
+ col_id=col_id, start_row=start_row,
+ rel_valid=rel_valid,
+ end_row=end_row, col_string=col_string)
+ # Extract the training part of data,
+ # which corresponds to rows outside of [start_row, end_row).
+ # Extract the validation part of data,
+ # which corresponds to rows inside of [start_row, end_row).
+ sql = """
+ drop view if exists {rel_train};
+ create temp view {rel_train} as
+ select {col_id}, {col_string} from {rel_source}
+ where {col_id} < {start_row}
+ or {col_id} >= {end_row};
+
+ drop view if exists {rel_valid};
+ create temp view {rel_valid} as
+ select {col_id}, {col_string} from {rel_source}
+ where {col_id} >= {start_row}
+ and {col_id} < {end_row}
+ """.format(**kwargs)
+ plpy.execute(sql)
+ return None
+# ------------------------------------------------------------------------------
+
+
+class CrossValidator(object):
+ """
+ Cross validation for estimator
+
+ Parameters
+ ==========
+ estimator : estimator function
+ The arguments to estimator are contained in args, e.g., estimator(**args)
+
+ predictor : predictor function
+ Arguments:
+
+ - schema_madlib: see args for details
+ - model_table: see args for details
+ - rel_valid: name of data table to be tested
+ - col_id: columns containing unique id for each data point
+ - output_table: table created for the testing results
+
+ scorer : string, either 'classification' or 'regression'
+ Indicate the scoring method to be used.
+
+ args : dict (recursive)
+ Contains all the arguments to run estimator and the data to be used for validation:
+
+ - source_table: the data table
+ - independent_varname: the column for features
+ - dependent_varname: the column for target
+ - schema_madlib: the schema where madlib is installed
+ - model_table: table created for the trained model
+
+ """
+ def __init__(self, estimator, predictor, scorer, args):
+ self._cv_args = _ValidationArgs(args)
+ self._estimator = estimator
+ self._predictor = predictor
+ self._scorer = scorer
+ self._target_col = None
+ self._features_col = None
+ self._set_data(**args)
+
+ def _set_data(self, source_table,
+ independent_varname,
+ dependent_varname, **kwargs):
+ self._col_data = _extract_data_cols(dependent_varname,
+ independent_varname)
+ self._col_id = unique_string(desp='col_id')
+ self._rel_copied = unique_string(desp='rel_copied')
+ self._row_num = num_samples(source_table)
+ y, x = _cv_copy_data(source_table, dependent_varname,
+ independent_varname,
+ self._rel_copied, self._col_id)
+ self._target_col, self._features_col = y, x
+
+ def _gen_split_data(self, n_folds):
+ rel_copied = self._rel_copied
+ col_data = [self._target_col, self._features_col]
+ col_id = self._col_id
+ row_num = self._row_num
+ SplitData = namedtuple('SplitData', 'rel_train, rel_valid')
+ for k in range(n_folds):
+ rel_train = unique_string(desp='cv_train_{0}'.format(k))
+ rel_valid = unique_string(desp='cv_valid_{0}'.format(k))
+ _cv_split_data(rel_copied, col_data, col_id, row_num,
+ rel_train, rel_valid, n_folds, k+1)
+ yield SplitData(rel_train=rel_train, rel_valid=rel_valid)
+
+ def _test_one_fold(self, split_data, sub_args):
+ col_id = self._col_id
+ estimator = self._estimator
+ predictor = self._predictor
+ scorer = self._scorer
+ rel_train = split_data.rel_train
+ rel_valid = split_data.rel_valid
+
+ args = self._cv_args.make_from(source_table=rel_train,
+ dependent_varname=self._target_col,
+ independent_varname=self._features_col,
+ **sub_args)
+ estimator(**args)
+
+ schema_madlib = args['schema_madlib']
+ output_table = unique_string(desp='output_table')
+ model_table = args['model_table']
+
+ predictor(schema_madlib, model_table,
+ rel_valid, col_id, output_table)
+ score = self._score(output_table, rel_valid, scorer)
+ plpy.execute("""
+ DROP TABLE IF EXISTS {model_table}, {model_table}_summary;
+ """.format(model_table=model_table))
+ plpy.execute("""
+ DROP TABLE IF EXISTS {output_table};
+ """.format(output_table=output_table))
+ return score
+
+ def _score(self, pred, orig, method):
+ target = self._target_col
+ col_id = self._col_id
+ if method == 'regression':
+ return plpy.execute(
+ """
+ SELECT
+ -avg(({target}-prediction)^2) AS accuracy
+ FROM {pred} JOIN {orig}
+ ON {pred}.{id} = {orig}.{id}
+ """.format(pred=pred,
+ orig=orig,
+ id=col_id,
+ target=target))[0]['accuracy']
+ elif method == 'classification':
+ return plpy.execute(
+ """
+ SELECT (1 - miss / total) AS accuracy
+ FROM
+ (
+ SELECT count(*)::float8 AS miss
+ FROM {pred} JOIN {orig}
+ ON {pred}.{id} = {orig}.{id}
+ WHERE prediction <> {target}) s,
+ (
+ SELECT count(*)::float8 AS total
+ FROM {orig}) r;
+ """.format(pred=pred,
+ orig=orig,
+ id=col_id,
+ target=target))[0]['accuracy']
+ else:
+ plpy.error("Cross Validation Error: invalid method value ({0})! "
+ "Need to be either classification "
+ "or regression!".format(method))
+
+ def validate(self, sub_args, n_folds=3):
+ """Returns the results of cross validation
+
+ Parameters
+ ==========
+ sub_args : dict (recursive)
+ Arguments to be validated. Multiple values are provided in a list, e.g.,
+
+ sub_args = {
+ 'params_dict':
+ {
+ 'lambda': [0.1, 1, 10],
+ 'epsilon': [0.1, 1, 10]
+ }
+ }
+
+ Before running estimator, sub_args_single is generated from sub_args replacing the lists with single value for each argument and args is updated recursively using sub_args_single.
+
+ n_folds : int, default=3
+ Number of folds. Must be at least 2
+
+ Returns
+ =======
+ validation_result : object
+ See class ValidationResult for more details
+ """
+ def _stats(nums):
+ n = len(nums)
+ # no need to check against 0 division
+ # because n_folds is larger than 1
+ a = sum(nums) / n
+ s = sqrt(sum([(x - a)**2 for x in nums]) / (n - 1))
+ return a, s
+
+ if not sub_args:
+ return []
+
+ cv_history = ValidationResult()
+ split_data = list(self._gen_split_data(n_folds))
+ for sa in _ValidationArgs.grid(sub_args):
+ _test = partial(self._test_one_fold, sub_args=sa)
+ scores = map(_test, split_data)
+ a, s = _stats(scores)
+ cv_history.add_one(mean=a, std=s, sub_args=sa)
+ return cv_history
\ No newline at end of file