// Path: src/main/java/vn/vitk/util/SparkContextFactory.java
// public class SparkContextFactory {
// 	private static volatile JavaSparkContext jsc = null; 
// 	
// 	private SparkContextFactory() {}
// 	
// 	public static JavaSparkContext create(String appName, String master) {
// 		if (jsc == null) {
// 			synchronized (SparkContextFactory.class) {
// 				if (jsc == null) {
// 					SparkConf sc = new SparkConf().setAppName(appName).setMaster(master);
// 					jsc = new JavaSparkContext(sc);
// 				}
// 			}
// 		}
// 		return jsc;
// 	}
// 
// 	public static JavaSparkContext create(String master) {
// 		if (jsc == null) {
// 			synchronized (SparkContextFactory.class) {
// 				if (jsc == null) {
// 					SparkConf sc = new SparkConf()
// 						.setAppName("vn.vitk")
// 						.setMaster(master);
// 					jsc = new JavaSparkContext(sc);
// 				}
// 			}
// 		}
// 		return jsc;
// 	}
// 	
// 	public static JavaSparkContext create() {
// 		if (jsc == null) {
// 			synchronized (SparkContextFactory.class) {
// 				if (jsc == null) {
// 					SparkConf sc = new SparkConf()
// 						.setAppName("vn.vitk")
// 						.setMaster("local[*]");
// 					jsc = new JavaSparkContext(sc);
// 				}
// 			}
// 		}
// 		return jsc;
// 	}
// }


import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.PosixParser;
import org.apache.spark.Accumulator;
import org.apache.spark.AccumulatorParam;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import scala.Tuple2;
import vn.vitk.util.SparkContextFactory;
import de.l3s.boilerpipe.BoilerpipeProcessingException;
import de.l3s.boilerpipe.extractors.ArticleExtractor;
package vn.vitk.tok;




/**
 * @author Phuong LE-HONG
 * <p>
 * Mar 11, 2016, 10:20:06 AM
 * <p>
 * Fast and simple reimplementation of <code>vnTokenizer</code> tool.
 */
public class Tokenizer implements Serializable {
	
	private static final long serialVersionUID = 216079852616487675L;
	
	private transient JavaSparkContext jsc;
	private Lexicon lexicon = new Lexicon();
	private Map<String, Pattern> patterns = new TreeMap<String, Pattern>();
	private TextNormalizationFunction normalizationFunction = new TextNormalizationFunction();
	private PhraseGraph graph = new PhraseGraph();
	private Bigrams bigram = null;
	private WhitespaceClassifier classifier = null;
	private Accumulator<List<WhitespaceContext>> contexts;
	private PipelineModel model = null;
	private Broadcast<Row[]> prediction = null;
	private int counter = 0;
	
	private boolean verbose = false;
	
	/**
	 * Creates a Vietnamese tokenizer.
	 * @param master
	 * @param lexiconFileName
	 * @param regexpFileName
	 */
	public Tokenizer(String master, String lexiconFileName, String regexpFileName) {
// Path: src/main/java/vn/vitk/util/SparkContextFactory.java
// public class SparkContextFactory {
// 	private static volatile JavaSparkContext jsc = null; 
// 	
// 	private SparkContextFactory() {}
// 	
// 	public static JavaSparkContext create(String appName, String master) {
// 		if (jsc == null) {
// 			synchronized (SparkContextFactory.class) {
// 				if (jsc == null) {
// 					SparkConf sc = new SparkConf().setAppName(appName).setMaster(master);
// 					jsc = new JavaSparkContext(sc);
// 				}
// 			}
// 		}
// 		return jsc;
// 	}
// 
// 	public static JavaSparkContext create(String master) {
// 		if (jsc == null) {
// 			synchronized (SparkContextFactory.class) {
// 				if (jsc == null) {
// 					SparkConf sc = new SparkConf()
// 						.setAppName("vn.vitk")
// 						.setMaster(master);
// 					jsc = new JavaSparkContext(sc);
// 				}
// 			}
// 		}
// 		return jsc;
// 	}
// 	
// 	public static JavaSparkContext create() {
// 		if (jsc == null) {
// 			synchronized (SparkContextFactory.class) {
// 				if (jsc == null) {
// 					SparkConf sc = new SparkConf()
// 						.setAppName("vn.vitk")
// 						.setMaster("local[*]");
// 					jsc = new JavaSparkContext(sc);
// 				}
// 			}
// 		}
// 		return jsc;
// 	}
// }


// Path: src/main/java/vn/vitk/tok/Tokenizer.java
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.PosixParser;
import org.apache.spark.Accumulator;
import org.apache.spark.AccumulatorParam;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import scala.Tuple2;
import vn.vitk.util.SparkContextFactory;
import de.l3s.boilerpipe.BoilerpipeProcessingException;
import de.l3s.boilerpipe.extractors.ArticleExtractor;

package vn.vitk.tok;




/**
 * @author Phuong LE-HONG
 * <p>
 * Mar 11, 2016, 10:20:06 AM
 * <p>
 * Fast and simple reimplementation of <code>vnTokenizer</code> tool.
 */
public class Tokenizer implements Serializable {
	
	private static final long serialVersionUID = 216079852616487675L;
	
	private transient JavaSparkContext jsc;
	private Lexicon lexicon = new Lexicon();
	private Map<String, Pattern> patterns = new TreeMap<String, Pattern>();
	private TextNormalizationFunction normalizationFunction = new TextNormalizationFunction();
	private PhraseGraph graph = new PhraseGraph();
	private Bigrams bigram = null;
	private WhitespaceClassifier classifier = null;
	private Accumulator<List<WhitespaceContext>> contexts;
	private PipelineModel model = null;
	private Broadcast<Row[]> prediction = null;
	private int counter = 0;
	
	private boolean verbose = false;
	
	/**
	 * Creates a Vietnamese tokenizer.
	 * @param master
	 * @param lexiconFileName
	 * @param regexpFileName
	 */
	public Tokenizer(String master, String lexiconFileName, String regexpFileName) {
		jsc = SparkContextFactory.create(master);