public final class DependencyGraph implements Serializable {
	
	private static final long serialVersionUID = -3230714950937126020L;
	
	private Sentence sentence;
	private Integer[] heads;
	private String[] labels;
	
	/**
	 * Creates a dependency graph.
	 * @param sentence
	 * @param heads
	 * @param labels
	 */
	public DependencyGraph(Sentence sentence, Integer[] heads, String[] labels) {
		this.sentence = sentence;
		this.heads = heads;
		this.labels = labels;
		if (sentence.length() != heads.length || heads.length != labels.length) {
			throw new IllegalArgumentException("Lengths do not match!");
		}
	}
	
	/**
	 * Creates a dependency graph.
	 * @param sentence
	 * @param arcs
	 */
	public DependencyGraph(Sentence sentence, List<Dependency> arcs) {
		this.sentence = sentence;
		int n = sentence.length();
		this.heads = new Integer[n];
		this.labels = new String[n];
		for (int i = 0; i < n; i++) {
			heads[i] = -1;
			labels[i] = "root";
		}
		// fill in 'heads' and 'labels'
		for (Dependency arc : arcs) {
			int u = arc.getHead();
			int v = arc.getDependent();
			heads[v] = u;
			labels[v] = arc.getType().substring(3); // "??-label" --> "label"
		}
	}
	/**
	 * Checks if the graph has arc (u, v) with any label.
	 * @param u
	 * @param v
	 * @return <code>true</code> or <code>false</code>
	 */
	public boolean hasArc(int u, int v) {
		if (u < 0 || u >= sentence.length() || v < 0 || v >= sentence.length()) {
			throw new IllegalArgumentException("Invalid index!");
		}
		return (heads[v] == u);
	}
	
	/**
	 * Checks the projectivity of this graph.
	 * @return <code>true</code> or <code>false</code>
	 */
	public boolean isProjective() {
		// TODO:
		return true;
	}
	
	/**
	 * Checks the connectivity of this graph.
	 * @return <code>true</code> or <code>false</code>
	 */
	public boolean isConnected() {
		// TODO:
		return true;
	}
	

	public Sentence getSentence() {
		return sentence;
	}
	
	public Integer[] getHeads() {
		return heads;
	}
	
	public String[] getLabels() {
		return labels;
	}
	
	/**
	 * Updates the dependency label at a position.
	 * @param position
	 * @param label
	 */
	public void setLabel(int position, String label) {
		labels[position] = label;
	}
	
	/**
	 * Updates the head at a position
	 * @param position
	 * @param head
	 */
	public void setHead(int position, int head) {
		heads[position] = head;
	}
	
	/**
	 * Gets the root position of the graph.
	 * @return the head position.
	 */
	public int getRoot() {
		for (int j = 0; j < heads.length; j++)
			if (heads[j] == 0)
				return j;
		return -1;
	}
	
	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder(512);
		int n = sentence.length();
		for (int j = 0; j < n; j++) {
			sb.append(j);
			sb.append('\t');
			sb.append(sentence.getToken(j));
			sb.append('\t');
			sb.append(sentence.getLemma(j));
			sb.append('\t');
			sb.append(sentence.getTag(j));
			sb.append('\t');
			sb.append(heads[j]);
			sb.append('\t');
			sb.append(labels[j]);
			sb.append('\n');
		}
		return sb.toString();
	}
	
	/**
	 * Generates a flattened list of dependency relations containing label(src-idx, tar-idx) tuples, separated by the whitespace character.
	 * @return a string
	 */
	public String dependencies() {
		StringBuilder sb = new StringBuilder();
		int n = sentence.length();
		for (int j = 0; j < n; j++) {
			if (heads[j] >= 0) {
				sb.append(labels[j]);
				sb.append('(');
				sb.append(sentence.getToken(heads[j]));
				sb.append('-');
				sb.append(heads[j]);
				sb.append(',');
				sb.append(sentence.getToken(j));
				sb.append('-');
				sb.append(j);
				sb.append(')');
				sb.append(' ');
			}
		}		
		return sb.toString().trim();
	}
} public class DependencyGraphReader {


	/**
	 * Reads a list of dependency graphs from a text file.
	 * @param fileName
	 * @param format either 'x' or 'u' (universal) 
	 * @return a list of dependency graphs.
	 */
	public static List<DependencyGraph> read(String fileName, char format) {
		if (format == 'x')
			return readCoNLLX(fileName);
		else if (format == 'u') 
			return readCoNLLU(fileName);
		return null;
	}
	
	/**
	 * Reads a list of dependency graphs from a text file.
	 * @param fileName
	 * @return a list of dependency graphs.
	 */
	private static List<DependencyGraph> readCoNLLX(String fileName) {
		List<DependencyGraph> graphs = new ArrayList<DependencyGraph>();
		BufferedReader br = null;
		try {
			br = new BufferedReader(new InputStreamReader(new FileInputStream(fileName), "UTF-8"));
			String line = null;
			List<String> tokens = new ArrayList<String>();
			tokens.add("ROOT");
			List<String> tags = new ArrayList<String>();
			tags.add("ROOT");
			List<Integer> heads = new ArrayList<Integer>();
			heads.add(-1);
			List<String> labels = new ArrayList<String>();
			labels.add("NULL");
			
			while ((line = br.readLine()) != null) {
				if (line.trim().length() == 0) {
					int n = tokens.size();
					Sentence sentence = new Sentence(tokens.toArray(new String[n]), tags.toArray(new String[n]));
					DependencyGraph graph = new DependencyGraph(sentence, heads.toArray(new Integer[n]), labels.toArray(new String[n]));
					graphs.add(graph);
					tokens = new ArrayList<String>();
					tokens.add("ROOT");
					tags = new ArrayList<String>();
					tags.add("ROOT");
					heads = new ArrayList<Integer>();
					heads.add(-1);
					labels = new ArrayList<String>();
					labels.add("NULL");
				} else {
					String[] parts = line.split("\\s+");
					if (parts.length != 10) {
						throw new IllegalArgumentException("Bad file format! " + line);
					}
					tokens.add(parts[1]);
					tags.add(parts[4]);
					heads.add(new Integer(parts[6]));
					labels.add(parts[7]);
				}
			}
			br.close();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				br.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return graphs;
	}
	
	/**
	 * Reads a list of dependency graphs from a text file.
	 * @param fileName CoNLL-U format.
	 * @return a list of dependency graphs.
	 */
	private static List<DependencyGraph> readCoNLLU(String fileName) {
		List<DependencyGraph> graphs = new ArrayList<DependencyGraph>();
		BufferedReader br = null;
		try {
			br = new BufferedReader(new InputStreamReader(new FileInputStream(fileName), "UTF-8"));
			String line = null;
			List<String> tokens = new LinkedList<String>();
			tokens.add("ROOT");
			List<String> lemmas = new LinkedList<String>();
			lemmas.add("ROOT");
			List<String> tags = new LinkedList<String>();
			tags.add("ROOT");
			List<Integer> heads = new LinkedList<Integer>();
			heads.add(-1);
			List<String> labels = new LinkedList<String>();
			labels.add("NULL");
			
			while ((line = br.readLine()) != null) {
				if (line.trim().length() == 0) {
					// a new graph
					int n = tokens.size();
					Sentence sentence = new Sentence(
							tokens.toArray(new String[n]),
							lemmas.toArray(new String[n]),
							tags.toArray(new String[n]));
					DependencyGraph graph = new DependencyGraph(sentence, heads.toArray(new Integer[n]), labels.toArray(new String[n]));
					graphs.add(graph);
					tokens = new LinkedList<String>();
					tokens.add("ROOT");
					lemmas = new LinkedList<String>();
					lemmas.add("ROOT");
					tags = new LinkedList<String>();
					tags.add("ROOT");
					heads = new LinkedList<Integer>();
					heads.add(-1);
					labels = new LinkedList<String>();
					labels.add("NULL");
				} else {
					// construct a graph
					String[] parts = line.split("\\s+");
					if (parts.length != 10) {
						throw new IllegalArgumentException("Bad file format! " + line);
					}
					tokens.add(parts[1]);// word form or punctuation symbol
					lemmas.add(parts[2]); // lemma
					tags.add(parts[3]); // universal PoS tag  
					heads.add(new Integer(parts[6])); // head of the current token
					labels.add(parts[7]); // universal dependency relation to the head (or root iff head==0)
				}
			}
			br.close();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				br.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return graphs;
	}
	
} public abstract class DependencyParser {
	
	public enum OutputFormat {
		TEXT,
		JSON,
	}
	
	/**
	 * Parses a sentence. 
	 * @param sentence
	 * @return a dependency graph
	 */
	public abstract DependencyGraph parse(Sentence sentence);
	
	
	/**
	 * Parse a list of sentences in a parallel fashion.
	 * @param jsc
	 * @param sentences
	 * @return a list of dependency graphs.
	 */
	public List<DependencyGraph> parse(JavaSparkContext jsc, List<Sentence> sentences) {
		JavaRDD<Sentence> jrdd = jsc.parallelize(sentences);
		return jrdd.map(new ParsingFunction()).collect();
	}
	
	/**
	 * Parses a list of PoS-tagged sentences, each on a line and writes the result to an output 
	 * file in a specified output format.
	 * @param jsc
	 * @param sentences
	 * @param outputFileName
	 * @param outuptFormat
	 */
	public void parse(JavaSparkContext jsc, List<String> sentences, String outputFileName, OutputFormat outputFormat) {
		JavaRDD<String> input = jsc.parallelize(sentences);
		JavaRDD<Sentence> sents = input.map(new TaggedLineToSentenceFunction());
		JavaRDD<DependencyGraph> graphs = sents.map(new ParsingFunction());
		JavaRDD<Row> rows = graphs.map(new Function<DependencyGraph, Row>() {
			private static final long serialVersionUID = -812004521983071103L;
			public Row call(DependencyGraph graph) {
				return RowFactory.create(graph.getSentence().toString(), graph.dependencies());
			}
		});
		StructType schema = new StructType(new StructField[]{
			new StructField("sentence", DataTypes.StringType, false, Metadata.empty()),	
			new StructField("dependency", DataTypes.StringType, false, Metadata.empty())
		});
		SQLContext sqlContext = new SQLContext(jsc);
		DataFrame df = sqlContext.createDataFrame(rows, schema);
		
		if (outputFormat == OutputFormat.TEXT)  
			df.select("dependency").write().text(outputFileName);
		else 
			df.repartition(1).write().json(outputFileName);
	}
	
	private final class TaggedLineToSentenceFunction implements Function<String, Sentence> {
		private static final long serialVersionUID = -5656057350076339025L;
		@Override
		public Sentence call(String line) throws Exception {
			String[] taggedTokens = line.trim().split("\\s+");
			int n = taggedTokens.length + 1;
			String[] tokens = new String[n];
			String[] tags = new String[n];
			tokens[0] = "ROOT";
			tags[0] = "ROOT";
			for (int j = 1; j < n; j++) {
				String[] parts = taggedTokens[j-1].split("/");
				if (parts.length == 2) {
					tokens[j] = parts[0];
					tags[j] = parts[1];
				} else {
					if (parts.length == 3) {
						tokens[j] = parts[0];
						tags[j] = parts[2];
					} else {
						System.err.println("Line error: " + line);
					}
				}
			}
			return new Sentence(tokens, tags);
		}
	}
	
	/**
	 * Parses all sentences in an input file, each on a line and writes the result to an output 
	 * file in a specified output format.
	 * @param jsc
	 * @param inputFileName
	 * @param outputFileName
	 * @param outputFormat
	 */
	public void parse(JavaSparkContext jsc, String inputFileName, String outputFileName, OutputFormat outputFormat) {
		List<String> sentences = jsc.textFile(inputFileName).collect();
		parse(jsc, sentences, outputFileName, outputFormat);
	}
	
	
	/**
	 * Parses all sentences in an input file, each on a line and writes the result to 
	 * the console window containing flattened dependency tuples.
	 * @param jsc
	 * @param inputFileName
	 */
	public void parse(JavaSparkContext jsc, String inputFileName) {
		List<String> sentences = jsc.textFile(inputFileName).collect();
		JavaRDD<String> input = jsc.parallelize(sentences);
		JavaRDD<Sentence> sents = input.map(new TaggedLineToSentenceFunction());
		JavaRDD<DependencyGraph> graphs = sents.map(new ParsingFunction());
		JavaRDD<String> rows = graphs.map(new Function<DependencyGraph, String>() {
			private static final long serialVersionUID = -6021310762521034121L;

			public String call(DependencyGraph graph) {
				return graph.dependencies();
			}
		});
		for (String s : rows.collect()) {
			System.out.println(s);
		}
	}
	
	/**
	 * Evaluates a parser on a manually parsed corpus.
	 * @param jsc
	 * @param graphs
	 * @return a list of dependency graphs.
	 */
	public void evaluate(JavaSparkContext jsc, List<DependencyGraph> graphs) {
		List<Sentence> sentences = new LinkedList<Sentence>();
		for (DependencyGraph graph : graphs) {
			sentences.add(graph.getSentence());
		}
		JavaRDD<Sentence> jrdd = jsc.parallelize(sentences);
		List<DependencyGraph> proposedGraphs = jrdd.map(new ParsingFunction()).collect();
		Evaluator evaluator = new Evaluator();
		evaluator.evaluate(graphs, proposedGraphs);
		System.out.println("UAS(token) = " + evaluator.getUASToken());
		System.out.println("UAS(sentence) = " + evaluator.getUASSentence());
		System.out.println("LAS(token) = " + evaluator.getLASToken());
		System.out.println("LAS(sentence) = " + evaluator.getLASSentence());
	}
	
	private final class ParsingFunction implements Function<Sentence, DependencyGraph> {
		private static final long serialVersionUID = -427735844309327738L;
		@Override
		public DependencyGraph call(Sentence sentence) throws Exception {
			return parse(sentence);
		}
	}
	
} public enum FeatureFrame {
	TAG,
	TAG_TOKEN,
	TAG_TOKEN_TYPE,
} public class TransitionBasedParserMLP extends DependencyParser implements Serializable {

	private static final long serialVersionUID = -2300120949325582286L;
	
	private MultilayerPerceptronClassificationModel classifier;
	private PipelineModel pipelineModel;
	private FeatureFrame featureFrame;
	private String[] transitionName;
	private final Map<String, Integer> featureMap;
	
	/**
	 * Creates a transition-based parser using a MLP transition classifier.
	 * @param jsc
	 * @param classifierFileName
	 * @param featureFrame
	 */
	public TransitionBasedParserMLP(JavaSparkContext jsc, String classifierFileName, FeatureFrame featureFrame) {
		this.featureFrame = featureFrame;
		this.classifier = TransitionClassifier.load(jsc, new Path(classifierFileName, "data").toString());
		this.pipelineModel = PipelineModel.load(new Path(classifierFileName, "pipelineModel").toString());
		this.transitionName = ((StringIndexerModel)pipelineModel.stages()[2]).labels();
		String[] features = ((CountVectorizerModel)(pipelineModel.stages()[1])).vocabulary();
		this.featureMap = new HashMap<String, Integer>();
		for (int j = 0; j < features.length; j++) {
			this.featureMap.put(features[j], j);
		}
		
	}

	/* (non-Javadoc)
	 * @see vn.vitk.dp.DependencyParser#parse(vn.vitk.dp.Sentence)
	 */
	@Override
	public DependencyGraph parse(Sentence sentence) {
		Stack<Integer> stack = new Stack<Integer>();
		
		Queue<Integer> queue = new LinkedList<Integer>();
		for (int i = 0; i < sentence.length(); i++) 
			queue.add(i);
		
		List<Dependency> arcs = new ArrayList<Dependency>();
		FeatureExtractor featureBuilder = new FeatureExtractor(featureFrame);
		// the first transition is always a SHIFT
		Configuration config = new Configuration(sentence, stack, queue);
		config = config.next("SH");
		
		while (!config.isFinal()) {
			List<String> featureList = featureBuilder.extract(config);
			String[] fs = new String[featureList.size()];
			for (int j = 0; j < featureList.size(); j++) {
				fs[j] = featureList.get(j).toLowerCase();
			}
			List<Tuple2<Integer, Double>> x = new LinkedList<Tuple2<Integer, Double>>();
			for (String f : fs) {
				Integer i = featureMap.get(f);
				if (i != null) {
					x.add(new Tuple2<Integer, Double>(i, 1.0));
				}
			}
			Vector features = Vectors.sparse(featureMap.size(), x);
			double label = classifier.predict(features);
			
			String transition = transitionName[(int)label];
			// add the arc if it is a LA or RA transition
			// 
			Integer u = stack.peek();
			Integer v = queue.peek();
			if (transition.startsWith("LA")) { 
				arcs.add(new Dependency(v, u, transition));
			} else if (transition.startsWith("RA")) {
				arcs.add(new Dependency(u, v, transition));
			} else if (transition.equals("RE") && !config.isReducible()) {
				System.err.println("Predicted transition is REDUCE while the current configuration is not reducible");
			}
			config = config.next(transition);
		}
		return new DependencyGraph(sentence, arcs);
	}
	
} public class TransitionClassifier implements Serializable {

	private static final long serialVersionUID = -8638888022254374959L;

	protected TransitionClassifierParams params;
	protected SQLContext sqlContext;
	protected boolean verbose = false;

	public TransitionClassifier() {
		this.params = new TransitionClassifierParams();
	}
	
	public TransitionClassifier(TransitionClassifierParams params) {
		this.params = params;
	}

	
	/**
	 * Trains a transition classifier on the data frame.
	 * @param jsc
	 * @param graphs
	 * @param featureFrame
	 * @param classifierFileName
	 * @param numHiddenUnits
	 * @return a transition classifier.
	 */
	public Transformer trainMLP(JavaSparkContext jsc,
			List<DependencyGraph> graphs, FeatureFrame featureFrame,
			String classifierFileName, int numHiddenUnits) {
		// create a SQLContext
		this.sqlContext = new SQLContext(jsc);
		// extract a data frame from these graphs
		DataFrame dataset = toDataFrame(jsc, graphs, featureFrame);
		
		// create a processing pipeline and fit it to the data frame
		Pipeline pipeline = createPipeline();
		PipelineModel pipelineModel = pipeline.fit(dataset);
		DataFrame trainingData = pipelineModel.transform(dataset);
		
		// cache the training data for better performance
		trainingData.cache();
		
		if (verbose) {
			trainingData.show(false);
		}
		
		// compute the number of different labels, which is the maximum element 
		// in the 'label' column.
		trainingData.registerTempTable("dfTable");
		Row row = sqlContext.sql("SELECT MAX(label) as maxValue from dfTable").first();
		int numLabels = (int)row.getDouble(0);
		numLabels++;
		
		int vocabSize = ((CountVectorizerModel)(pipelineModel.stages()[1])).getVocabSize();
		
		// default is a two-layer MLP
		int[] layers = {vocabSize, numLabels};
		// if user specify a hidden layer, use a 3-layer MLP:
		if (numHiddenUnits > 0) {
			layers = new int[3];
			layers[0] = vocabSize;
			layers[1] = numHiddenUnits;
			layers[2] = numLabels;
		}
		MultilayerPerceptronClassifier classifier = new MultilayerPerceptronClassifier()
			.setLayers(layers)
			.setBlockSize(128)
			.setSeed(1234L)
			.setTol((Double)params.getOrDefault(params.getTolerance()))
			.setMaxIter((Integer)params.getOrDefault(params.getMaxIter()));
		MultilayerPerceptronClassificationModel model = classifier.fit(trainingData);
		
		// compute precision on the training data
		//
		DataFrame result = model.transform(trainingData);
		DataFrame predictionAndLabel = result.select("prediction", "label");
		MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator().setMetricName("precision");
		if (verbose) {
			System.out.println("N = " + trainingData.count());
			System.out.println("D = " + vocabSize);
			System.out.println("K = " + numLabels);
			System.out.println("H = " + numHiddenUnits);
			System.out.println("training precision = " + evaluator.evaluate(predictionAndLabel));
		}
		
		// save the trained MLP to a file
		//
		String classifierPath = new Path(classifierFileName, "data").toString();
		jsc.parallelize(Arrays.asList(model), 1).saveAsObjectFile(classifierPath);
		// save the pipeline model to sub-directory "pipelineModel"
		// 
		try {
			String pipelinePath = new Path(classifierFileName, "pipelineModel").toString(); 
			pipelineModel.write().overwrite().save(pipelinePath);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return model;
	}
	
	/**
	 * Converts a dependency graph to a data frame.
	 * @param jsc
	 * @param graph
	 * @param featureFrame
	 * @return a data frame
	 */
	public DataFrame toDataFrame(JavaSparkContext jsc, DependencyGraph graph, FeatureFrame featureFrame) {
		List<ParsingContext> list = TransitionDecoder.decode(graph, featureFrame);
		JavaRDD<ParsingContext> javaRDD = jsc.parallelize(list);
		return sqlContext.createDataFrame(javaRDD, ParsingContext.class);
	}

	/**
	 * Converts a list of dependency graphs to a data frame.
	 * @param jsc 
	 * @param graphs
	 * @param featureFrame
	 * @return a data frame
	 */
	private DataFrame toDataFrame(JavaSparkContext jsc, List<DependencyGraph> graphs, FeatureFrame featureFrame) {
		List<ParsingContext> list = new ArrayList<ParsingContext>();
		for (DependencyGraph graph : graphs) {
			List<ParsingContext> xy = TransitionDecoder.decode(graph, featureFrame);
			list.addAll(xy);
		}
		JavaRDD<ParsingContext> javaRDD = jsc.parallelize(list);
		return sqlContext.createDataFrame(javaRDD, ParsingContext.class);
	}
	
	/**
	 * Set the verbose mode.
	 * @param verbose
	 */
	public void setVerbose(boolean verbose) {
		this.verbose = verbose;
	}
	
	/**
	 * Creates a processing pipeline.
	 * @return a pipeline
	 */
	protected Pipeline createPipeline() {
		Tokenizer tokenizer = new Tokenizer()
			.setInputCol("text")
			.setOutputCol("tokens");
		CountVectorizer countVectorizer = new CountVectorizer()
			.setInputCol("tokens")
			.setOutputCol("features")
			.setMinDF((Double)params.getOrDefault(params.getMinFF()))
			.setVocabSize((Integer)params.getOrDefault(params.getNumFeatures()));  
		StringIndexer transitionIndexer = new StringIndexer()
			.setInputCol("transition")
			.setOutputCol("label");
		
		Pipeline pipeline = new Pipeline().setStages(new PipelineStage[]{tokenizer, countVectorizer, transitionIndexer});
		return pipeline;
	}

	/**
	 * Loads a MLP transition classifier from a file, which is pre-trained.
	 * @param jsc
	 * @param classifierFileName
	 */
	public static MultilayerPerceptronClassificationModel load(JavaSparkContext jsc, String classifierFileName) {
		Object object = jsc.objectFile(classifierFileName).first();
		return ((MultilayerPerceptronClassificationModel)object);
	}
	
} public class TransitionClassifierParams extends JavaParams {
	private static final long serialVersionUID = -3862425558891374277L;
	private String uid = null;
	private DoubleParam minFF = null;
	private IntParam numFeatures = null;
	private DoubleParam regParam = null;
	private IntParam maxIter = null;
	private DoubleParam tolerance = null;

	/**
	 * Creates default parameters.
	 */
	public TransitionClassifierParams() {
		minFF = new DoubleParam(this, "minFF", "min feature frequency", ParamValidators.gt(0));
		setDefault(minFF, 2.0);
		
		numFeatures = new IntParam(this, "numFeatures", "number of features used in feature hashing", ParamValidators.gt(0));
		setDefault(numFeatures, 1000);
		
		regParam = new DoubleParam(this, "regParam", "regularization parameter", ParamValidators.gtEq(0d));
		setDefault(regParam, 0.0);
		
		maxIter = new IntParam(this, "maxIter", "max number of iterations", ParamValidators.gt(0));
		setDefault(maxIter, 100);
		
		tolerance = new DoubleParam(this, "tolerance", "convergence tolerance of iterations", ParamValidators.gt(0));
		setDefault(tolerance, 1E-5);
	}

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

	/* (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 = "mlpParams" + "_" + ruid.substring(n-12, n);
		}
		return uid;
	}
	/**
	 * Set the minimal feature frequency, either an absolute integer value or 
	 * a real percentage value.
	 * @param value
	 * @return this object
	 */
	public TransitionClassifierParams setMinFF(double value) {
		set(minFF, value);
		return this;
	}
	
	/**
	 * Gets the minimal feature frequency parameter.
	 * @return the feature frequency parameter. 
	 */
	public DoubleParam getMinFF() {
		return minFF;
	}

	/**
	 * Set the number of features used in the feature hashing trick.
	 * @param value
	 * @return this object.
	 */
	public TransitionClassifierParams setNumFeatures(int value) {
		set(numFeatures, value);
		return this;
	}
	
	/**
	 * Gets the number of features.
	 * @return the number of feature parameter.
	 */
	public IntParam getNumFeatures() {
		return numFeatures;
	}
	
	/**
	 * Set the regularization value.
	 * @param value
	 * @return this object
	 */
	public TransitionClassifierParams setRegParam(double value) {
		set(regParam, value);
		return this;
	}
	
	/**
	 * Gets the regularization parameter.
	 * @return the regularization parameter.
	 */
	public DoubleParam getRegParam() {
		return regParam;
	}
	
	/**
	 * Set the max iterations in training.
	 * @param value
	 * @return this object.
	 */
	public TransitionClassifierParams setMaxIter(int value) {
		set(maxIter, value);
		return this;
	}
	
	/**
	 * Gets the max iteration parameter.
	 * @return the max iteration parameter.
	 */
	public IntParam getMaxIter() {
		return maxIter;
	}
	
	/**
	 * Set the convergence tolerance.
	 * @param value
	 * @return this object
	 */
	public TransitionClassifierParams setTolerance(double value) {
		set(tolerance, value);
		return this;
	}
	
	/**
	 * Gets the tolerance parameter.
	 * @return the tolerance parameter.
	 */
	public DoubleParam getTolerance() {
		return tolerance;
	}
} public class CMMParams extends JavaParams {
	private static final long serialVersionUID = -3862425558891374277L;
	private String uid = null;
	private DoubleParam minFF = null;
	private IntParam numFeatures = null;
	private DoubleParam regParam = null;
	private IntParam maxIter = null;
	private DoubleParam tolerance = null;
	private IntParam markovOrder = null;

	/**
	 * Creates default parameters.
	 */
	public CMMParams() {
		minFF = new DoubleParam(this, "minFF", "min feature frequency", ParamValidators.gt(0));
		setDefault(minFF, 2.0);
		
		numFeatures = new IntParam(this, "numFeatures", "number of features used in feature hashing", ParamValidators.gt(0));
		setDefault(numFeatures, 1000);
		
		regParam = new DoubleParam(this, "regParam", "regularization parameter", ParamValidators.gtEq(0d));
		setDefault(regParam, 0.0);
		
		maxIter = new IntParam(this, "maxIter", "max number of iterations", ParamValidators.gt(0));
		setDefault(maxIter, 100);
		
		tolerance = new DoubleParam(this, "tolerance", "convergence tolerance of iterations", ParamValidators.gt(0));
		setDefault(tolerance, 1E-5);
		
		markovOrder = new IntParam(this, "markovOrder", "Markov order of the model", ParamValidators.gt(0));
		setDefault(markovOrder, 1);
	}

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

	/* (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 = "cmmParams" + "_" + ruid.substring(n-12, n);
		}
		return uid;
	}
	/**
	 * Set the minimal feature frequency, either an absolute integer value or 
	 * a real percentage value.
	 * @param value
	 * @return this object
	 */
	public CMMParams setMinFF(double value) {
		set(minFF, value);
		return this;
	}
	
	/**
	 * Gets the minimal feature frequency parameter.
	 * @return the feature frequency parameter. 
	 */
	public DoubleParam getMinFF() {
		return minFF;
	}

	/**
	 * Set the number of features used in the feature hashing trick.
	 * @param value
	 * @return this object.
	 */
	public CMMParams setNumFeatures(int value) {
		set(numFeatures, value);
		return this;
	}
	
	/**
	 * Gets the number of features.
	 * @return the number of feature parameter.
	 */
	public IntParam getNumFeatures() {
		return numFeatures;
	}
	
	/**
	 * Set the regularization value.
	 * @param value
	 * @return this object
	 */
	public CMMParams setRegParam(double value) {
		set(regParam, value);
		return this;
	}
	
	/**
	 * Gets the regularization parameter.
	 * @return the regularization parameter.
	 */
	public DoubleParam getRegParam() {
		return regParam;
	}
	
	/**
	 * Set the max iterations in training.
	 * @param value
	 * @return this object.
	 */
	public CMMParams setMaxIter(int value) {
		set(maxIter, value);
		return this;
	}
	
	/**
	 * Gets the max iteration parameter.
	 * @return the max iteration parameter.
	 */
	public IntParam getMaxIter() {
		return maxIter;
	}
	
	/**
	 * Set the convergence tolerance.
	 * @param value
	 * @return this object
	 */
	public CMMParams setTolerance(double value) {
		set(tolerance, value);
		return this;
	}
	
	/**
	 * Gets the tolerance parameter.
	 * @return the tolerance parameter.
	 */
	public DoubleParam getTolerance() {
		return tolerance;
	}
	
	/**
	 * Set the Markov order (currently, 1 or 2 -- May 2016).
	 * @param value
	 * @return this object.
	 */
	public CMMParams setMarkovOrder(int value) {
		set(markovOrder, value);
		return this;
	}

	/**
	 * Gets the Markov order parameter.
	 * @return the Markov order parameter.
	 */
	public IntParam getMarkovOrder() {
		return markovOrder;
	}

} public class Tagger implements Serializable {
	
	private static final long serialVersionUID = 8061440373376898771L;
	private transient JavaSparkContext jsc;
	private CMMModel cmmModel;
	
	private boolean verbose = false;
	
	public enum OutputFormat {
		TEXT,
		JSON,
		PARQUET
	}
	
	enum TaggerMode {
		TRAIN,
		TAG
	}
	
	
	/**
	 * Creates a tagger which runs on a stand-alone machine.
	 */
	public Tagger(JavaSparkContext jsc) {
		this.jsc = jsc;
	}
	
	/**
	 * Trains a tagger. The <code>inputFileName</code> contains tagged sentences
	 * in a simple format: "He/P kicks/V a/D ball/N ./.". After training, the
	 * tagger model is saved to a file <code>modelFileName</code>. The
	 * parameters used in training is given in an argument.
	 * 
	 * @param inputFileName
	 * @param modelFileName
	 * @param params
	 * @return a {@link CMMModel}
	 */
	public CMMModel train(String inputFileName, String modelFileName, CMMParams params) {
		JavaRDD<String> lines = jsc.textFile(inputFileName);
		cmmModel = train(lines.collect(), modelFileName, params);
		return cmmModel;
	}
	
	/**
	 * Trains a tagger with data specified in a data frame. The data frame has 
	 * two columns, one column "sentence" contains a word sequence, and the other column "partOfSpeech" 
	 * contains the corresponding tag sequence. Each row of the data frame specifies a tagged sequence
	 * in the training set.
	 * @param dataset
	 * @param modelFileName
	 * @param params
	 * @return a {@link CMMModel}
	 */
	public CMMModel train(DataFrame dataset, String modelFileName, CMMParams params) {
		CMM cmm = new CMM(params).setVerbose(verbose);
		cmmModel = cmm.fit(dataset);
		try {
			cmmModel.write().overwrite().save(modelFileName);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return cmmModel;
	}
	
	/**
	 * Trains a tagger. Training data are tagged sequences stored in an input
	 * file of a simple format, each sequence in a line. After training, the
	 * tagger is saved to a file. The parameters used in training is given in 
	 * an argument.
	 * 
	 * @param taggedSentences
	 * @param modelFileName
	 * @param params
	 * @return a {@link CMMModel}
	 */
	public CMMModel train(List<String> taggedSentences, String modelFileName, CMMParams params) {
		DataFrame dataset = createDataFrame(taggedSentences);
		return train(dataset, modelFileName, params);
	}

	/**
	 * Creates a data frame from a list of tagged sentences.
	 * @param taggedSentences
	 * @return a data frame of two columns: "sentence" and "partOfSpeech".
	 */
	public DataFrame createDataFrame(List<String> taggedSentences) {
		List<String> wordSequences = new LinkedList<String>();
		List<String> tagSequences = new LinkedList<String>();
		for (String taggedSentence : taggedSentences) {
			StringBuilder wordBuf = new StringBuilder();
			StringBuilder tagBuf = new StringBuilder();
			String[] tokens = taggedSentence.split("\\s+");
			for (String token : tokens) {
				String[] parts = token.split("/");
				if (parts.length == 2) {
					wordBuf.append(parts[0]);
					wordBuf.append(' ');
					tagBuf.append(parts[1]);
					tagBuf.append(' ');
				} else { // this token is "///"  
					wordBuf.append('/');
					wordBuf.append(' ');
					tagBuf.append('/');
					tagBuf.append(' ');
				}
			}
			wordSequences.add(wordBuf.toString().trim());
			tagSequences.add(tagBuf.toString().trim());
		}
		if (verbose) {
			System.out.println("Number of sentences = " + wordSequences.size());
		}
		List<Row> rows = new LinkedList<Row>();
		for (int i = 0; i < wordSequences.size(); i++) {
			rows.add(RowFactory.create(wordSequences.get(i), tagSequences.get(i)));
		}
		JavaRDD<Row> jrdd = jsc.parallelize(rows);
		StructType schema = new StructType(new StructField[]{
				new StructField("sentence", DataTypes.StringType, false, Metadata.empty()),
				new StructField("partOfSpeech", DataTypes.StringType, false, Metadata.empty())
			});
			
		return new SQLContext(jsc).createDataFrame(jrdd, schema);
	}
	
	/**
	 * Loads a {@link CMMModel} from a model file.
	 * @param modelFileName
	 * @return this object.
	 */
	public Tagger load(String modelFileName) {
		this.cmmModel = CMMModel.load(modelFileName);
		return this;
	}
	
	
	/**
	 * Tags a list of sequences and returns a list of tag sequences.
	 * @param sentences
	 * @return a list of tagged sequences.
	 */
	public List<String> tag(List<String> sentences) {
		List<Row> rows = new LinkedList<Row>();
		for (String sentence : sentences) {
			rows.add(RowFactory.create(sentence));
		}
		StructType schema = new StructType(new StructField[]{
			new StructField("sentence", DataTypes.StringType, false, Metadata.empty())	
		});
		SQLContext sqlContext = new SQLContext(jsc);
		DataFrame input = sqlContext.createDataFrame(rows, schema);
		if (cmmModel != null) {
			DataFrame output = cmmModel.transform(input).repartition(1);
			return output.javaRDD().map(new RowToStringFunction(1)).collect();
		} else {
			System.err.println("Tagging model is null. You need to create or load a model first.");
			return null;
		}
	}
	
	/**
	 * Tags a data frame containing a column named 'sentence'.
	 * @param input
	 * @param outputFileName
	 * @param outputFormat
	 */
	public void tag(DataFrame input, String outputFileName, OutputFormat outputFormat) {
		long tic = System.currentTimeMillis();
		long duration = 0;
		if (cmmModel != null) {
			DataFrame output = cmmModel.transform(input).repartition(1);
			duration = System.currentTimeMillis() - tic;
			switch (outputFormat) {
			case JSON:
				output.write().json(outputFileName);
				break;
			case PARQUET:
				output.write().parquet(outputFileName);
				break;
			case TEXT:
				toTaggedSentence(output).repartition(1).saveAsTextFile(outputFileName);
//				output.select("prediction").write().text(outputFileName);
				break;
			}
		} else {
			System.err.println("Tagging model is null. You need to create or load a model first.");
		}
		if (verbose) {
			long n = input.count();
			System.out.println(" Number of sentences = " + n);
			System.out.println("  Total tagging time = " + duration + " milliseconds.");
			System.out.println("Average tagging time = " + ((float)duration) / n + " milliseconds.");
		}
	}
	
	/**
	 * Tags a list of sequences and writes the result to an output file with a
	 * desired output format.
	 * 
	 * @param sentences
	 * @param outputFileName
	 * @param outputFormat
	 */
	public void tag(List<String> sentences, String outputFileName, OutputFormat outputFormat) {
		List<Row> rows = new LinkedList<Row>();
		for (String sentence : sentences) {
			rows.add(RowFactory.create(sentence));
		}
		StructType schema = new StructType(new StructField[]{
			new StructField("sentence", DataTypes.StringType, false, Metadata.empty())	
		});
		SQLContext sqlContext = new SQLContext(jsc);
		DataFrame input = sqlContext.createDataFrame(rows, schema);
		tag(input, outputFileName, outputFormat);
	}
	
	/**
	 * Tags a distributed list of sentences and writes the result to an output file with 
	 * a desired output format.
	 * @param sentences
	 * @param outputFileName
	 * @param outputFormat
	 */
	public void tag(JavaRDD<Row> sentences, String outputFileName, OutputFormat outputFormat) {
		StructType schema = new StructType(new StructField[]{
			new StructField("sentence", DataTypes.StringType, false, Metadata.empty())	
		});
		SQLContext sqlContext = new SQLContext(jsc);
		DataFrame input = sqlContext.createDataFrame(sentences, schema);
		tag(input, outputFileName, outputFormat);
	}
	
	/**
	 * Tags a text file, each sentence in a line and writes the result to an output file 
	 * with a desired output format.
	 * @param inputFileName
	 * @param outputFileName
	 * @param outputFormat
	 */
	public void tag(String inputFileName, String outputFileName, OutputFormat outputFormat) {
		List<String> sentences = jsc.textFile(inputFileName).collect();
		tag(sentences, outputFileName, outputFormat);
	}
	
	/**
	 * Tags a text file, each sentence in a line and writes the result to the console.
	 * @param inputFileName
	 */
	public void tag(String inputFileName) {
		List<String> sentences = jsc.textFile(inputFileName).collect();
		List<String> output = tag(sentences);
		for (int i = 0; i < sentences.size(); i++) {
			StringBuilder sb = new StringBuilder(64);
			String words[] = sentences.get(i).split("\\s+");
			String tags[] = output.get(i).split("\\s+");
			for (int j = 0; j < words.length; j++) {
				sb.append(words[j]);
				sb.append('/');
				sb.append(tags[j]);
				sb.append(' ');
			}
			System.out.println(sb.toString().trim());
		}
	}
	
	private JavaRDD<String> toTaggedSentence(DataFrame output) {
		return output.javaRDD().map(new Function<Row, String>() {
			private static final long serialVersionUID = 4208643510231783579L;
			@Override
			public String call(Row row) throws Exception {
				String[] tokens = row.getString(0).trim().split("\\s+");
				String[] tags = row.getString(1).trim().split("\\s+");
				if (tokens.length != tags.length) {
					System.err.println("Incompatible lengths!");
					return null;
				}
				StringBuilder sb = new StringBuilder(64);
				for (int j = 0; j < tokens.length; j++) {
					sb.append(tokens[j]);
					sb.append('/');
					sb.append(tags[j]);
					sb.append(' ');
				}
				return sb.toString().trim();
			}
		});
	}
	
	/**
	 * Evaluates the accuracy of a CMM model on a data frame on tagged sentences.
	 * @param dataset
	 * @return evaluation measures.
	 */
	public float[] evaluate(DataFrame dataset) {
		List<String> correctSequences = dataset.javaRDD().map(new RowToStringFunction(1)).collect();
		long beginTime = System.currentTimeMillis();
		DataFrame output = cmmModel.transform(dataset);
		long endTime = System.currentTimeMillis();
		if (verbose) {
			System.out.println(" Number of sentences = " + correctSequences.size());
			long duration = (endTime - beginTime);
			System.out.println("  Total tagging time = " + duration + " ms.");
			System.out.println("Average tagging time = " + ((float)duration) / correctSequences.size() + " ms.");
		}
		List<String> automaticSequences = output.javaRDD().map(new RowToStringFunction(1)).collect();
		return Evaluator.evaluate(automaticSequences, correctSequences);
	}
	
	/**
	 * Evaluates the tagger on a manually tagged data file.
	 * @param inputFileName
	 * @return scores
	 */
	public float[] evaluate(String inputFileName) {
		JavaRDD<String> lines = jsc.textFile(inputFileName);
		DataFrame dataset = createDataFrame(lines.collect());
		return evaluate(dataset);
	}
	
	/**
	 * Evaluates the accuracy of a CMM model on a list of tagged sentences. 
	 * @param taggedSentences
	 * @return evaluation measures.
	 */
	public float[] evaluate(List<String> taggedSentences) {
		DataFrame dataset = createDataFrame(taggedSentences);
		return evaluate(dataset);
	}
	
	private class RowToStringFunction implements Function<Row, String> {
		private static final long serialVersionUID = -2245906041132281238L;
		private int columnIndex = 0;
		
		RowToStringFunction(int columnIndex) {
			this.columnIndex = columnIndex;
		}
		@Override
		public String call(Row row) throws Exception {
			return row.getString(columnIndex);
		}
	}
	
	/**
	 * Set the verbose mode.
	 * @param verbose
	 * @return this object
	 */
	public Tagger setVerbose(boolean verbose) {
		this.verbose = verbose;
		return this;
	}

	void test(String modelFileName, TaggerMode mode) {
		String[] taggedSentences = {
				"tôi/P ăn/V quả/Nc chuối/N to/A ./.",
				"tôi/Np đá/V quả/Nc bóng/N ./.",
				"tôi/Np ăn/V quả/Nc bóng/N to/A ./.",
				"vải/N hoa/N bụng/N cóc/N",
				"tôi/P cóc/R sợ/A" 
		};
		if (mode == TaggerMode.TRAIN) {
			train(Arrays.asList(taggedSentences), modelFileName, 
					new CMMParams().setNumFeatures(30));
		} else if (mode == TaggerMode.TAG){
			load(modelFileName);
			String[] sentences = {
				"tôi ăn quả chuối to .",
				"tôi đá quả bóng .",
				"tôi ăn quả bóng to .", 
				"vải hoa bụng cóc",
				"tôi cóc sợ" 
			};
			tag(Arrays.asList(sentences), "dat/tag/out", OutputFormat.JSON);
			evaluate(Arrays.asList(taggedSentences));
		}
	}
	
	void test(String inputFileName, int numFeatures, String modelFileName, TaggerMode mode) {
		if (mode == TaggerMode.TRAIN) {
			CMMParams params = new CMMParams()
				.setMaxIter(600)
				.setNumFeatures(numFeatures);
			train(inputFileName, modelFileName, params);
		}
		else if (mode == TaggerMode.TAG) {
			load(modelFileName);
			List<String> taggedSentences = jsc.textFile(inputFileName).collect();
			evaluate(taggedSentences);
		}
	}
	
	void testRandomSplit(String inputFileName, int numFeatures, String modelFileName) {
		CMMParams params = new CMMParams()
			.setMaxIter(600)
			.setRegParam(1E-6)
			.setMarkovOrder(2)
			.setNumFeatures(numFeatures);
		
		JavaRDD<String> lines = jsc.textFile(inputFileName);
		DataFrame dataset = createDataFrame(lines.collect());
		DataFrame[] splits = dataset.randomSplit(new double[]{0.9, 0.1}); 
		DataFrame trainingData = splits[0];
		System.out.println("Number of training sequences = " + trainingData.count());
		DataFrame testData = splits[1];
		System.out.println("Number of test sequences = " + testData.count());
		// train and save a model on the training data
		cmmModel = train(trainingData, modelFileName, params);
		// test the model on the test data
		System.out.println("Test accuracy:");
		evaluate(testData); 
		// test the model on the training data
		System.out.println("Training accuracy:");
		evaluate(trainingData);
	}
	
	/**
	 * For internal test only.
	 * @param args
	 */
	public static void main(String[] args) {
		CorpusPack cp = new CorpusPack(Language.VIETNAMESE);
		String modelFileName = cp.taggerModelFileName();
		
		JavaSparkContext jsc = SparkContextFactory.create();
		Tagger tagger = new Tagger(jsc).setVerbose(true);
		// 1. Toy dataset
		tagger.test(modelFileName, TaggerMode.TRAIN);
		tagger.test(modelFileName, TaggerMode.TAG);
		
		// 2. VTB, randomSplit (90%, 10%)
//		String corpusFileName = cp.taggerCorpusFileName();
//		tagger.testRandomSplit(corpusFileName, 160000, modelFileName);
	}
} 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);
		lexicon = new Lexicon().load(lexiconFileName);
		if (verbose) 
			System.out.println("#(nodes of the lexicon) = " + lexicon.numNodes());
		
		List<String> lines = jsc.textFile(regexpFileName).collect();
		for (String line : lines) {
			line = line.trim();
			if (!line.startsWith("#")) { // ignore comment lines
				String[] s = line.split("\\s+");
				if (s.length == 2) {
					patterns.put(s[0], Pattern.compile(s[1]));
				}
			}
		}
	}
	
	/**
	 * Creates a Vietnamese tokenizer.
	 * @param master
	 * @param lexiconFileName
	 * @param regexpFileName
	 * @param bigramFileName
	 */
	public Tokenizer(String master, String lexiconFileName, String regexpFileName, String bigramFileName) {
		this(master, lexiconFileName, regexpFileName);
		bigram = new Bigrams(bigramFileName);
	}

	public Tokenizer(String master, String lexiconFileName, String regexpFileName, String whitespaceModelFileName, boolean lr) {
		this(master, lexiconFileName, regexpFileName);
		classifier = new WhitespaceClassifier(lexicon, patterns);
		model = classifier.load(whitespaceModelFileName);
		contexts = jsc.accumulator(new LinkedList<WhitespaceContext>(), new WhitespaceContextAccumulatorParam());
	}
	
	/**
	 * Reads the content of a text file to get lines.
	 * @param fileName
	 * @return a RDD of text lines.
	 */
	public JavaRDD<String> readTextFile(String fileName) {
		JavaRDD<String> input = jsc.textFile(fileName);
		return input.map(normalizationFunction);
	}
	
	/**
	 * Tokenizes a RDD of text lines and return a RDD of result.
	 * @param input
	 * @return a RDD of tokenized text lines.
	 */
	public JavaRDD<String> tokenize(JavaRDD<String> input) {
		if (verbose) {
			// print some basic statistic about the input, including 
			// max line length, min line length, average line length in syllables
			JavaRDD<Integer> wordCount = input.map(new Function<String, Integer>() {
				private static final long serialVersionUID = 7214093453452927565L;
				@Override
				public Integer call(String line) throws Exception {
					return line.split("\\s+").length;
				}
				
			});
			Comparator<Integer> comp = new IntegerComparator();
			System.out.println("Max line length (in syllables) = " + wordCount.max(comp));
			System.out.println("Min line length (in syllables) = " + wordCount.min(comp));
			float totalCount = wordCount.reduce(new Function2<Integer, Integer, Integer>() {
				private static final long serialVersionUID = 1L;
				@Override
				public Integer call(Integer v1, Integer v2) throws Exception {
					return v1 + v2;
				}
			});
			System.out.println("Avg line length (in syllables) = " + (totalCount) / input.count());
		}
		
		JavaRDD<String> output = null;
		if (classifier == null) {
			// use phrase graph approach (shortest paths and bigram model)
			// to segment phrases
			output = input.map(new SegmentationFunction());
		} else {
			// use logistic regression approach to segment phrases
			JavaRDD<String> s = input.map(new SegmentationFunction());
			// make sure that the preceding lazy computation has been evaluated
			// so that whitespace contexts have been properly accumulated
			System.out.println("Number of text lines = " + s.count());
			System.out.println("Number of contexts = " + contexts.value().size());
			// use whitespace classification approach (logistic regresion model)
			JavaRDD<WhitespaceContext> jrdd = jsc.parallelize(contexts.value());
			DataFrame df0 = (new SQLContext(jsc)).createDataFrame(jrdd, WhitespaceContext.class);
			DataFrame df1 = model.transform(df0);
			prediction = jsc.broadcast(df1.select("prediction").collect());
			if (df1.count() > 0) {
				output = s.map(new WhitespaceClassificationFunction());
			}
			else { 
				System.err.println("Empty data frame!");
			}
		}
		if (verbose) {
			// print number of non-space characters of the input and output dataset
			System.out.println("#(non-space characters of input) = " + numCharacters(input));
			if (output != null) {
				System.out.println("#(non-space characters of output) = " + numCharacters(output));
			}
		}
		return output;
	}
	
	
	/**
	 * Tokenizes a line. 
	 * @param line a line of text
	 * @return a result text string
	 */
	public String tokenizeOneLine(String line) {
		List<String> list = new ArrayList<String>();
		list.add(line);
		JavaRDD<String> input = jsc.parallelize(list);
		JavaRDD<String> output = tokenize(input);
		return output.first();
	}
	
	/**
	 * Tokenizes a text file and returns a list of tokens.
	 * @param fileName
	 * @return a list of tokens.
	 */
	public List<String> tokenize(String fileName) {
		JavaRDD<String> input = readTextFile(fileName);
		JavaRDD<String> output = tokenize(input);
		return output.collect();
	}
	
	/**
	 * Tokenizes a text file and saves the result to an output directory by using 
	 * Spark save as text file utility.
	 * @param inputFileName
	 * @param outputDirectory
	 */
	public void tokenize(String inputFileName, String outputDirectory) {
		JavaRDD<String> input = readTextFile(inputFileName);
		JavaRDD<String> output = tokenize(input);
		output.saveAsTextFile(outputDirectory);
	}
	
	
	/**
	 * Tokenizes a text file and writes the result to a writer. 
	 * @param inputFileName
	 * @param writer
	 * @return tokenization result.
	 */
	public List<String> tokenize(String inputFileName, PrintWriter writer) {
		JavaRDD<String> input = readTextFile(inputFileName);
		JavaRDD<String> output = tokenize(input);
		List<String> lines = output.collect();
		for (String line : lines) {
			writer.write(line);
			writer.write('\n');
		}
		writer.flush();
		return lines;
	}
	
	/**
	 * Tokenizes a text specified in an URL and writes the result to 
	 * a writer.
	 * @param url
	 * @param writer
	 */
	public List<String> tokenize(URL url, PrintWriter writer) {
		try {
			System.out.println("Extracting the text content of the URL...");
			String text = ArticleExtractor.INSTANCE.getText(new InputStreamReader(url.openStream(), "UTF-8"));
			if (verbose) {
				System.out.println("URL text content:");
				System.out.println(text);
			}
			System.out.println("Tokenizing the content...");
			JavaRDD<String> input = jsc.parallelize(Arrays.asList(text.split("\\n+")));
			JavaRDD<String> output = tokenize(input.map(normalizationFunction));
			List<String> lines = output.collect();
			for (String line : lines) {
				writer.write(line);
				writer.write('\n');
			}
			writer.flush();
			return lines;
		} catch (BoilerpipeProcessingException e) {
			e.printStackTrace();
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}
	
	
	/**
	 * Counts the number of non-space characters in this data set. This utility method 
	 * is used to check the tokenization result.
	 * @param lines
	 * @return number of characters
	 */
	int numCharacters(JavaRDD<String> lines) {
		JavaRDD<Integer> lengths = lines.map(new Function<String, Integer>() {
			private static final long serialVersionUID = -2189399343462982586L;
			@Override
			public Integer call(String line) throws Exception {
				line = line.replaceAll("[\\s_]+", "");
				return line.length();
			}
		});
		return lengths.reduce(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = -8438072946884289401L;

			@Override
			public Integer call(Integer e0, Integer e1) throws Exception {
				return e0 + e1;
			}
		});
	}
	
	/**
	 * Adds more words to the lexicon.
	 * @param words a list of words.
	 */
	public void addWords(List<String> words) {
		addWords(words.toArray(new String[words.size()]));
	}
	
	/**
	 * Adds more words to the lexicon.
	 * @param words a string of words.
	 */
	public void addWords(String[] words) {
		for (String word : words)
			lexicon.addWord(word);
	}
	
	class SegmentationFunction implements Function<String, String> {
		private static final long serialVersionUID = 7206190528458789338L;

		@Override
		public String call(String sentence) throws Exception {
			Iterable<Tuple2<String, String>> tokens = segment(sentence);
			StringBuilder sb = new StringBuilder(64);
			for (Tuple2<String, String> token : tokens) {
				String content = token._2();
				if (!token._1().equals("phrase")) {
					content = content.replace(' ', '_');
				}
				sb.append(content);
				sb.append(' ');
			}
			return sb.toString().trim();
		}
		
		/**
		 * Segments a sentence, return an iterable of (token, type).
		 * @param sentence
		 * @return an iterable of (token, type) pairs.
		 */
		private Iterable<Tuple2<String, String>> segment(String sentence) {
			List<Tuple2<String, String>> tokens = new ArrayList<Tuple2<String, String>>();
			sentence = sentence.trim();
			if (sentence.length() == 0) {
				return tokens;
			}
			String s = sentence;
			while (true) {
				int maxLen = 0;
				String nextToken = "";
				String type = "";
				// greedy search for the longest pattern
				// from the beginning of 's'
				for (String patternName : patterns.keySet()) {
					Pattern pattern = patterns.get(patternName);
					Matcher matcher = pattern.matcher(s);
					if (matcher.lookingAt()) {
						int len = matcher.end() - matcher.start();
						if (maxLen < len) {
							maxLen = len;
							nextToken = matcher.group();
							type = patternName;
						}
					}
				}
				nextToken = nextToken.trim();
				if (nextToken.length() > 0) {
					s = s.substring(maxLen).trim();
					if (type.contains("name") && s.length() > 0) {
						Tuple2<String, String> tup = processName(nextToken, s);
						if (tup._1().length() != nextToken.length()) {
							nextToken = tup._1();
							s = tup._2();
							type = "word";
						}
						tokens.add(new Tuple2<String, String>(type, nextToken));
					} else if (type.contains("unit") && s.length() > 0) {
						Tuple2<String, String> tup = processUnit(nextToken, s);
						if (tup._1().length() > nextToken.length()) {
							nextToken = tup._1();
							s = tup._2();
							type = "unit";
						}
						tokens.add(new Tuple2<String, String>(type, nextToken));
					} else if (type.contains("phrase")) {
						if (nextToken.indexOf(' ') > 0) { // multi-syllabic phrase
							if (classifier != null) {
								contexts.add(classifier.makeContexts(nextToken));
								tokens.add(new Tuple2<String, String>("phrase", "[" + nextToken + "]"));
							} else {
								// segment the phrase using a phrase graph  
								List<String> words = tokenizePhrase(nextToken);
								if (words != null) {
									for (int i = 0; i < words.size(); i++) {
										tokens.add(new Tuple2<String, String>("word", words.get(i)));
									}
								} else {
									System.out.println("Error when tokenizing phrase: " + nextToken);
								}
							}
						} else { // mono-syllabic phrase, don't need to tokenize the phrase
							tokens.add(new Tuple2<String, String>("word", nextToken));
						}
					}  else { // next token is not a name, an unit or a phrase
						tokens.add(new Tuple2<String, String>(type, nextToken));
					}
				} else {
					if (s.trim().length() > 0) {
						System.out.println("Unprocessed substring: " + s);
					}
					break;
				}
				if (s.length() == 0) {
					break;
				}
			}
			return tokens;
		}
		
		private Tuple2<String, String> processName(String currentToken, String s) {
			// If this is a name pattern, we process it further to capture 2 cases: 
			//
			// 1. It should be merged with the next syllable, like in "Thủ tướng" or "Bộ Giáo dục." where 
			// the name pattern captures only the first part "Thủ" or "Bộ Giáo".
			// We try to combine the last syllable of the current token with the first token of s 
			// to see whether they may form a word or not, note that the first token of s may contain 
			// delimiters (like "dục." in the example above); we therefore need to remove them 
			// beforehand if necessary.
			int j = s.indexOf(' ');
			String nextSyllable = (j > 0) ? s.substring(0, j) : s;
			// s can either be "dục" or "dục.", find the last alphabetic character of s
			// so as to leave the non-alphabetic characters out.
			int u = nextSyllable.length();
			while (u > 0 && !Character.isAlphabetic(nextSyllable.charAt(--u)));
			nextSyllable = nextSyllable.substring(0, u+1);
			
			int k = currentToken.lastIndexOf(' ');
			String lastSyllable = (k > 0) ? currentToken.substring(k+1) : currentToken;
			String nextTokenPrefix = (k > 0) ? currentToken.substring(0, k+1) : "";  
			String w = lastSyllable.toLowerCase() + ' ' + nextSyllable;
			if (lexicon.hasWord(w)) {
				currentToken = nextTokenPrefix + lastSyllable + ' ' + nextSyllable;
				s = s.substring(nextSyllable.length()).trim();
			}
			// 2. It should be divided into two parts if the first syllable of the name 
			// is a name prefix like "Ông", "Bà", "Anh", "Em", etc.
			j = currentToken.indexOf(' ');
			if (j > 0) {
				String firstSyllable = currentToken.substring(0, j);
				Matcher matcher = patterns.get("prefix").matcher(firstSyllable);
				if (matcher.matches()) {
					StringBuilder sb = new StringBuilder(currentToken.substring(j+1));
					sb.append(' ');
					sb.append(s);
					s = sb.toString();
					currentToken = firstSyllable;
				}
			}
			return new Tuple2<String, String>(currentToken, s);
		}
		
		private Tuple2<String, String> processUnit(String currentToken, String s) {
			// "[đồng/đô] [la Mỹ...]" => [đồng/đô la] [Mỹ...] 
			// "[đồng/cổ phiếu] [...]"
			String lastSyllable = currentToken.substring(currentToken.indexOf('/') + 1);
			int j = s.indexOf(' ');
			String nextSyllable = (j > 0) ? s.substring(0, j) : s;
			// s can either be "phiếu" or "phiếu.", find the last alphabetic character of s
			// so as to leave the non-alphabetic characters out.
			int u = nextSyllable.length();
			while (u > 0 && !Character.isAlphabetic(nextSyllable.charAt(--u)));
			nextSyllable = nextSyllable.substring(0, u+1);
			
			if (lexicon.hasWord(lastSyllable + ' ' + nextSyllable)) {
				currentToken = currentToken + ' ' + nextSyllable;
				s = s.substring(nextSyllable.length()).trim();
			}
			return new Tuple2<String, String>(currentToken, s);
		}
		
		/**
		 * Tokenizes a phrase.
		 * @param phrase
		 * @return a list of tokens.
		 */
		private List<String> tokenizePhrase(String phrase) {
			graph.makeGraph(phrase);
			List<LinkedList<Integer>> paths = graph.shortestPaths();
			if (paths.size() > 0) {
//				// print out overlap groups
//				List<Tuple2<Integer, Integer>> ambiguities = graph.overlaps();
//				for (Tuple2<Integer, Integer> tup : ambiguities) {
//					System.out.println(graph.words(tup));
//				}
				LinkedList<Integer> selectedPath = paths.get(paths.size()-1);
				if (bigram != null) {
					int best = graph.select(paths);
					selectedPath = paths.get(best);
				}
				return graph.words(selectedPath);
			}
			if (verbose) {
				System.out.println("Cannot tokenize the following phrase: [" + phrase +"]");
			}
			return null;
		}
		
	}
	
	class PhraseGraph implements Serializable {
		private static final long serialVersionUID = 6761055345566557524L;
		private String[] syllables;
		private int n;
		
		/**
		 * For each vertex v, we store a list of vertices u where (u, v) is an edge
		 * of the graph. This is used to recursively search for all paths on the graph.  
		 */
		private Map<Integer, LinkedList<Integer>> edges = new HashMap<Integer, LinkedList<Integer>>();

		void makeGraph(String phrase) {
			edges.clear();
			syllables = phrase.split("\\s+");
			n = syllables.length;
			if (n > 128) {
				System.out.println("WARNING: Phrase too long (>= 128 syllables), tokenization may be slow...");
				System.out.println(phrase);
			}
			for (int j = 0; j <= n; j++) {
				edges.put(j, new LinkedList<Integer>());
			}
			for (int i = 0; i < n; i++) {
				String token = syllables[i];
				int j = i;
				while (j < n) {
					if (lexicon.hasWord(token)) {
						edges.get(j+1).add(i);
					}
					j++;
					if (j < n) {
						token = token + ' ' + syllables[j];
					}
				}
			}
			// make sure that the graph is connected by adding adjacent 
			// edges if necessary
			for (int i = n; i > 0; i--) {
				if (edges.get(i).size() == 0) { // i cannot reach by any previous node
					edges.get(i).add(i-1);
				}
			}
		}

		/**
		 * Finds all shortest paths from the first node to the last node
		 * of this graph. 
		 * @return a list of paths, each path is a linked list of vertices.
		 */
		public List<LinkedList<Integer>> shortestPaths() {
			Dijkstra dijkstra = new Dijkstra(edges);
			List<LinkedList<Integer>> allPaths = dijkstra.shortestPaths();
			if (verbose) {
				if (allPaths.size() > 16) {
					StringBuilder phrase = new StringBuilder();
					for (String syllable : syllables) {
						phrase.append(syllable);
						phrase.append(' ');
					}
					System.out.printf("This phrase is too ambiguous, giving %d shortest paths!\n\t%s\n", 
							allPaths.size(), phrase.toString().trim());
				}
			}
			return allPaths;
		}
		
		/**
		 * Gets a list of words specified by a given path.
		 * @param path
		 * @return a list of words.
		 */
		public List<String> words(LinkedList<Integer> path) {
			int m = path.size();
			if (m <= 1) 
				return null;
			Integer[] a = path.toArray(new Integer[m]);
			StringBuilder[] tok = new StringBuilder[m-1];
			int i;
			for (int j = 0; j < m-1; j++) {
				// get the token from a[j] to a[j+1] (exclusive)
				tok[j] = new StringBuilder();
				i = a[j];
				tok[j].append(syllables[i]);
				for (int k = a[j]+1; k < a[j+1]; k++) {
					tok[j].append(' ');
					tok[j].append(syllables[k]);
				}
			}
			List<String> result = new LinkedList<String>();
			for (StringBuilder sb : tok) {
				result.add(sb.toString());
			}
			return result;
		}
		
		/**
		 * Gets a sub-sequence of syllables from a segment marking 
		 * the beginning and the end positions.
		 * @param segment
		 * @return
		 */
		public String words(Tuple2<Integer, Integer> segment) {
			StringBuilder sb = new StringBuilder();
			for (int i = segment._1(); i < segment._2(); i++) {
				sb.append(syllables[i]);
				sb.append(' ');
			}
			return sb.toString().trim();
		}
		
		/**
		 * Finds all overlap groups of syllables of this graph. A syllable group is overlap if
		 * it defines multiple shortest paths, e.g. 4 nodes and two paths: [0, 1, 3] or [0, 2, 3]. 
		 * @return a list of syllable groups, each is a tuple of begin and end position.
		 */
		public List<Tuple2<Integer, Integer>> overlaps() {
			List<Tuple2<Integer, Integer>> result = new ArrayList<Tuple2<Integer, Integer>>();
			if (n >= 4) {
				int i = n-1;
				while (i >= 3) {
					if (edges.get(i).contains(i-1) && edges.get(i).contains(i-2)
							&& edges.get(i-1).contains(i-3) && edges.get(i-2).contains(i-3)) {
						result.add(new Tuple2<Integer, Integer>(i-3, i));
						i = i - 4;
					} else {
						i = i - 1;
					}
				}
			}
			return result;
		}
		
		@Override
		public String toString() {
			return edges.toString();
		}
		
		/**
		 * Selects the most likely segmentation from a list 
		 * of different segmentations.
		 * @param paths
		 * @return
		 */
		public int select(List<LinkedList<Integer>> paths) {
			if (bigram != null) {
				int maxIdx = 0;
				double maxVal = Double.MIN_VALUE;
				// find the maximum of log probabilities of segmentations
				for (int j = 0; j < paths.size(); j++) {
					LinkedList<Integer> path = paths.get(j);
					List<String> words = words(path);
					words.add(0, "<s>");
					words.add("</s>");
					double p = 0d;
					for (int w = 1; w < words.size(); w++)
						p += bigram.logConditionalProb(words.get(w-1), words.get(w));
					if (p > maxVal) {
						maxVal = p;
						maxIdx = j;
					}
				}
				return maxIdx;
			}
			return 0;
		}
	}

	class TextNormalizationFunction implements Function<String, String> {
		private static final long serialVersionUID = 5727433453096616457L;
		private Map<String, String> vowels = new HashMap<String, String>();
		private Pattern pattern = Pattern.compile("[hklmst][yỳýỷỹỵ]");  
		private Map<Character, Character> ymap = new HashMap<Character, Character>();

		public TextNormalizationFunction() {
			// initialize the vowel map
			vowels.put("òa", "oà");
			vowels.put("óa", "oá");
			vowels.put("ỏa", "oả");
			vowels.put("õa", "oã");
			vowels.put("ọa", "oạ");
			vowels.put("òe", "oè");
			vowels.put("óe", "oé");
			vowels.put("ỏe", "oẻ");
			vowels.put("õe", "oẽ");
			vowels.put("ọe", "oẹ");
			vowels.put("ùy", "uỳ");
			vowels.put("úy", "uý");
			vowels.put("ủy", "uỷ");
			vowels.put("ũy", "uỹ");
			vowels.put("ụy", "uỵ");
			// initialize the y map
			ymap.put('y', 'y');
			ymap.put('ỳ', 'ì');
			ymap.put('ý', 'í');
			ymap.put('ỷ', 'ỉ');
			ymap.put('ỹ', 'ĩ');
			ymap.put('ỵ', 'ị');
		}
		
		@Override
		public String call(String phrase) throws Exception {
			// normalize all the vowels of the phrase
			for (String u : vowels.keySet()) {
				String v = vowels.get(u);
				phrase = phrase.replace(u, v);
			}
			// normalize the i/y by converting 'y' to 'i'  
			// if 'y' goes after consonants "h, k, l, m, s, t".
			StringBuilder sb = new StringBuilder(phrase);
			Matcher matcher = pattern.matcher(phrase);
			while (matcher.find()) {
				int idx = matcher.start() + 1;
				sb = sb.replace(idx, matcher.end(), String.valueOf(ymap.get(sb.charAt(idx))));
			}
			return sb.toString();
		}
	}
	
	
	class WhitespaceContextAccumulatorParam implements AccumulatorParam<List<WhitespaceContext>> {

		private static final long serialVersionUID = -8140972589714419311L;
		
		@Override
		public List<WhitespaceContext> addInPlace(List<WhitespaceContext> l1,
				List<WhitespaceContext> l2) {
			l1.addAll(l2);
			return l1;
		}

		@Override
		public List<WhitespaceContext> zero(List<WhitespaceContext> list) {
			return new LinkedList<WhitespaceContext>();
		}

		@Override
		public List<WhitespaceContext> addAccumulator(
				List<WhitespaceContext> l1, List<WhitespaceContext> l2) {
			l1.addAll(l2);
			return l1;
		}

	}
	
	class WhitespaceClassificationFunction implements Function<String, String> {
		private static final long serialVersionUID = -8841083728315408933L;
		private Pattern phrase = Pattern.compile("\\[[\\p{Ll}\\s]+\\]");
		
		@Override
		public String call(String sentence) throws Exception {
			StringBuilder r = new StringBuilder(sentence);
			Matcher matcher = phrase.matcher(sentence);
			while (matcher.find()) {
				int u = matcher.start();
				int v = matcher.end();
				String s = sentence.substring(u+1, v-1);
				int i = 0;
				for (int j = 0; j < s.length(); j++)
					if (s.charAt(j) == ' ') i++;
				String[] syllables = s.split("\\s+");
				StringBuilder sb = new StringBuilder(s.length());
				for (int j = 0; j < i; j++) {
					sb.append(syllables[j]);
					if (prediction.value()[counter + j].getDouble(0) == 0) {
						sb.append(' ');
					} else {
						sb.append('_');
					}
				}
				sb.append(syllables[syllables.length-1]);
				counter += i;
				r.replace(u+1, v-1, sb.toString());
			}
			return r.toString();
		}
		
	}
	
	class IntegerComparator implements Comparator<Integer>, Serializable {
		
		private static final long serialVersionUID = 3285846060042662009L;

		@Override
		public int compare(Integer i, Integer j) {
			return i - j;
		}
		
	}
	
	public void setVerbose(boolean verbose) {
		this.verbose = verbose;
	}
	
	/**
	 * For internal test only.
	 * @param args
	 * @throws MalformedURLException 
	 */
	public static void main(String[] args) throws MalformedURLException {
		String master = "local[*]";
		String inputFileName = "";
		String outputFileName = "";
		String url = "";
		
		Options options = new Options();
		options.addOption("m", true, "master");
		options.addOption("i", true, "input file name)");
		options.addOption("o", true, "output file name");
		options.addOption("u", true, "input URL");
		options.addOption("v", false, "verbose");
		options.addOption("s", false, "whitespace classification");
		CommandLineParser parser = new PosixParser();
		CommandLine cm;
		try {
			cm = parser.parse(options, args);
			if (cm.hasOption("m")) {
				master = cm.getOptionValue("m");
			}
			String dataFolder = "/export/dat/tok";
			Tokenizer tokenizer = null;
			if (cm.hasOption("s")) {
				// use whitespace classification
				tokenizer = new Tokenizer(master, dataFolder + "/lexicon.xml", 
						dataFolder + "/regexp.txt", dataFolder + "/whitespace.model", true);
			} else {
				// use a bigram model
				tokenizer = new Tokenizer(master, dataFolder + "/lexicon.xml", 
						dataFolder + "/regexp.txt", dataFolder + "/syllables2M.arpa");
			}
			if (cm.hasOption("v")) {
				tokenizer.setVerbose(true);
			}
			if (cm.hasOption("i")) {
				inputFileName = cm.getOptionValue("i");
			}
			if (cm.hasOption("u")) {
				url = cm.getOptionValue("u");
			}
			if (inputFileName.length() == 0 && url.length() == 0) {
				System.err.println("Either an input file or an URL must be provided!");
				System.exit(1);
			} else if (inputFileName.length() > 0) {
				if (cm.hasOption("o")) {
					outputFileName = cm.getOptionValue("o");
					tokenizer.tokenize(inputFileName, outputFileName);
				} else {
					try {
						PrintWriter writer = new PrintWriter(new OutputStreamWriter(System.out, "UTF-8"));
						tokenizer.tokenize(inputFileName, writer);
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
				}
			} else {
				try {
					PrintWriter writer = null;
					if (cm.hasOption("o")) {
						outputFileName = cm.getOptionValue("o");
						writer = new PrintWriter(new OutputStreamWriter(new FileOutputStream(outputFileName), "UTF-8"));
						tokenizer.tokenize(new URL(url), writer);
						writer.close();
					} else {
						writer = new PrintWriter(new OutputStreamWriter(System.out, "UTF-8"));
						tokenizer.tokenize(new URL(url), writer);
					}
				} catch (UnsupportedEncodingException | FileNotFoundException e) {
					e.printStackTrace();
				}
			}
		} catch (ParseException e) {
			e.printStackTrace();
		}
		System.out.println("Done.");
	}
} 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.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.List;
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.api.java.JavaSparkContext;
import vn.vitk.dep.DependencyGraph;
import vn.vitk.dep.DependencyGraphReader;
import vn.vitk.dep.DependencyParser;
import vn.vitk.dep.FeatureFrame;
import vn.vitk.dep.TransitionBasedParserMLP;
import vn.vitk.dep.TransitionClassifier;
import vn.vitk.dep.TransitionClassifierParams;
import vn.vitk.tag.CMMParams;
import vn.vitk.tag.Tagger;
import vn.vitk.tok.Tokenizer;
import vn.vitk.util.SparkContextFactory;
						PrintWriter writer = null;
						if (cm.hasOption("o")) {
							outputFileName = cm.getOptionValue("o");
							writer = new PrintWriter(new OutputStreamWriter(new FileOutputStream(outputFileName), "UTF-8"));
							tokenizer.tokenize(new URL(url), writer);
							writer.close();
						} else {
							writer = new PrintWriter(new OutputStreamWriter(System.out, "UTF-8"));
							tokenizer.tokenize(new URL(url), writer);
						}
					} catch (UnsupportedEncodingException | FileNotFoundException | MalformedURLException e) {
						e.printStackTrace();
					}
				}
			} else if (tool.equalsIgnoreCase("tag")) {
				// 2. part-of-speech tagging
				// 
				String dataFolder = "/export/dat/tag";
				String action = "tag";
				String cmm = dataFolder + "/vi/cmm";
				
				Tagger tagger = new Tagger(jsc);
				
				if (cm.hasOption("v")) {
					tagger.setVerbose(true);
				}
				
				if (cm.hasOption("cmm")) {
					cmm = cm.getOptionValue("cmm");
				}
				
				if (cm.hasOption("i")) {
					inputFileName = cm.getOptionValue("i");
				} else {
					System.err.println("You need to provide an input filename!");
					System.exit(1);
				}
				
				if (cm.hasOption("a")) {
					action = cm.getOptionValue("a");
				}
				if (action.equalsIgnoreCase("train")) {
					int numFeatures = -1;
					double regParam = 0d;
					if (cm.hasOption("dim")) {
						numFeatures = Integer.parseInt(cm.getOptionValue("dim"));
					}
					if (cm.hasOption("reg")) {
						regParam = Double.parseDouble(cm.getOptionValue("reg"));
					}
					if (numFeatures > 0) {
						CMMParams params = new CMMParams()
							.setMarkovOrder(2)
							.setMaxIter(600).setNumFeatures(numFeatures)
							.setRegParam(regParam);
						tagger.train(inputFileName, cmm, params);
					} else {
						System.err.println("Number of features must be positive!");
					}
				} else if (action.equalsIgnoreCase("tag")) {
					tagger.load(cmm);
					if (cm.hasOption("o")) {
						outputFileName = cm.getOptionValue("o");
						tagger.tag(inputFileName, outputFileName, Tagger.OutputFormat.TEXT);
					} else {
						tagger.tag(inputFileName);
					}
					
				} else if (action.equalsIgnoreCase("eval")) {
					tagger.load(cmm);
					tagger.evaluate(inputFileName);
				} else {
					System.err.println("You need to provide an action: 'tag', 'eval' or 'train'!");
				}
			} else if (tool.equalsIgnoreCase("dep")) {
				// 3. dependency parsing 
				//
				String dataFolder = "/export/dat/dep/vi";
				String action = "parse";
				boolean verbose = false;
				
				if (cm.hasOption("v")) {
					verbose = true;
				}
				
				if (language.equals("en")) {
					dataFolder = "/export/dat/dep/en";
				}
				String mlp = dataFolder + "/mlp";
				
				if (cm.hasOption("mlp")) {
					mlp = cm.getOptionValue("mlp");
				}
				
				if (cm.hasOption("i")) {
					inputFileName = cm.getOptionValue("i");
				} else {
					System.err.println("You need to provide an input filename!");
					System.exit(1);
				}
				char format = language.equalsIgnoreCase("en") ? 'u' : 'x';
				
				if (cm.hasOption("a")) {
					action = cm.getOptionValue("a");
				}
				

				if (action.equalsIgnoreCase("train")) {
					int numFeatures = -1;
					int numHiddenUnits = 0;

					if (cm.hasOption("dim")) {
						numFeatures = Integer.parseInt(cm.getOptionValue("dim"));
					}
					if (cm.hasOption("hid")) {
						numHiddenUnits = Integer.parseInt(cm.getOptionValue("hid"));
					}
					
					if (numFeatures > 0) {
						TransitionClassifierParams params = new TransitionClassifierParams().setNumFeatures(numFeatures);
						TransitionClassifier classifier = new TransitionClassifier(params);