// Path: src/main/java/vn/vitk/util/Constants.java
// public interface Constants {
// 	public static final String REGEXP_FILE = "/export/dat/tok/regexp.txt";
// }


import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.ml.Estimator;
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.feature.CountVectorizer;
import org.apache.spark.ml.feature.CountVectorizerModel;
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.ml.feature.Tokenizer;
import org.apache.spark.ml.param.ParamMap;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.optimization.LBFGS;
import org.apache.spark.mllib.optimization.LogisticGradient;
import org.apache.spark.mllib.optimization.SquaredL2Updater;
import org.apache.spark.mllib.util.MLUtils;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;
import vn.vitk.util.Constants;
package vn.vitk.tag;




/**
 * @author Phuong LE-HONG
 *         <p>
 *         May 9, 2016, 6:09:08 PM
 *         <p>
 *         Conditional Markov model for sequence labeling. This models is also
 *         called Maximum-Entropy Markov model (MEMM).
 *         <p>
 *         The input of this estimator is a data frame containing two columns: "featureStrings"
 *         and "label".
 * 
 */
public class CMM extends Estimator<CMMModel> {

	private static final long serialVersionUID = -6366026106299743238L;
	private String uid = null;
	private int numLabels = 0;
	private CMMParams params;
	
	private boolean verbose = false;
	
	/**
	 * Constructs a Conditional Markov Model for sequence labeling with default 
	 * parameters for use in training.
	 */
	public CMM() {
		params = new CMMParams();
	}
	
	/**
	 * Creates a Conditioanl Markov model for sequence labeling with given 
	 * parameters for use in training.
	 * @param params
	 */
	public CMM(CMMParams params) {
		this.params = params;
	}
	
	/* (non-Javadoc)
	 * @see org.apache.spark.ml.util.Identifiable#uid()
	 */
	@Override
	public String uid() {
		if (uid == null) {
			String ruid = UUID.randomUUID().toString();
			int n = ruid.length();
			uid = "cmm" + "_" + ruid.substring(n-12, n);
		}
		return uid;
	}

	/* (non-Javadoc)
	 * @see org.apache.spark.ml.Estimator#copy(org.apache.spark.ml.param.ParamMap)
	 */
	@Override
	public Estimator<CMMModel> copy(ParamMap extra) {
		return defaultCopy(extra);
	}

	/* (non-Javadoc)
	 * @see org.apache.spark.ml.Estimator#fit(org.apache.spark.sql.DataFrame)
	 */
	@Override
	public CMMModel fit(DataFrame dataset) {
		MarkovOrder order = MarkovOrder.values()[(Integer)params.getOrDefault(params.getMarkovOrder())-1];
// Path: src/main/java/vn/vitk/util/Constants.java
// public interface Constants {
// 	public static final String REGEXP_FILE = "/export/dat/tok/regexp.txt";
// }


// Path: src/main/java/vn/vitk/tag/CMM.java
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.UUID;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.ml.Estimator;
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.feature.CountVectorizer;
import org.apache.spark.ml.feature.CountVectorizerModel;
import org.apache.spark.ml.feature.StringIndexer;
import org.apache.spark.ml.feature.Tokenizer;
import org.apache.spark.ml.param.ParamMap;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.optimization.LBFGS;
import org.apache.spark.mllib.optimization.LogisticGradient;
import org.apache.spark.mllib.optimization.SquaredL2Updater;
import org.apache.spark.mllib.util.MLUtils;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;
import vn.vitk.util.Constants;

package vn.vitk.tag;




/**
 * @author Phuong LE-HONG
 *         <p>
 *         May 9, 2016, 6:09:08 PM
 *         <p>
 *         Conditional Markov model for sequence labeling. This models is also
 *         called Maximum-Entropy Markov model (MEMM).
 *         <p>
 *         The input of this estimator is a data frame containing two columns: "featureStrings"
 *         and "label".
 * 
 */
public class CMM extends Estimator<CMMModel> {

	private static final long serialVersionUID = -6366026106299743238L;
	private String uid = null;
	private int numLabels = 0;
	private CMMParams params;
	
	private boolean verbose = false;
	
	/**
	 * Constructs a Conditional Markov Model for sequence labeling with default 
	 * parameters for use in training.
	 */
	public CMM() {
		params = new CMMParams();
	}
	
	/**
	 * Creates a Conditioanl Markov model for sequence labeling with given 
	 * parameters for use in training.
	 * @param params
	 */
	public CMM(CMMParams params) {
		this.params = params;
	}
	
	/* (non-Javadoc)
	 * @see org.apache.spark.ml.util.Identifiable#uid()
	 */
	@Override
	public String uid() {
		if (uid == null) {
			String ruid = UUID.randomUUID().toString();
			int n = ruid.length();
			uid = "cmm" + "_" + ruid.substring(n-12, n);
		}
		return uid;
	}

	/* (non-Javadoc)
	 * @see org.apache.spark.ml.Estimator#copy(org.apache.spark.ml.param.ParamMap)
	 */
	@Override
	public Estimator<CMMModel> copy(ParamMap extra) {
		return defaultCopy(extra);
	}

	/* (non-Javadoc)
	 * @see org.apache.spark.ml.Estimator#fit(org.apache.spark.sql.DataFrame)
	 */
	@Override
	public CMMModel fit(DataFrame dataset) {
		MarkovOrder order = MarkovOrder.values()[(Integer)params.getOrDefault(params.getMarkovOrder())-1];
		ContextExtractor contextExtractor = new ContextExtractor(order, Constants.REGEXP_FILE);