package org.talend.recipeProcessor;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.api.java.Row;
import org.talend.recipeProcessor.spark.Concat;
import org.talend.recipeProcessor.spark.Sort;
import org.talend.recipeProcessor.spark.SparkSqlOperation;
import org.talend.recipeProcessor.spark.UpperCase;

public class Spark {

	public static void main(String[] args) throws IOException {
		SparkConf conf = new SparkConf().setAppName("test").setMaster("local");
		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> people = sc
				.textFile("/home/stef/talend/test_files/people.json");
		//
		// Model model = new Model();
		// model.table = "people";
		// model.columns = Arrays.asList("name", "age", "dept");
		//
		String asked = "";
		InputStreamReader isr = new InputStreamReader(System.in);
		BufferedReader br = new BufferedReader(isr);

		// JavaRDD<String> step1 = new UpperCase("name").apply(people);
		// System.out.println(step1.collect());
		//
		// JavaRDD<String> step2 = new Concat(Arrays.asList("name", "dept"))
		// .apply(step1);
		// System.out.println(step2.collect());

		while (asked != null && !asked.equals("q")) {
			if (!asked.equals("")) {
				System.out.println(" - " + asked);

				SparkSqlOperation operation = null;
				if (asked.equals("upper")) {
					operation = new UpperCase("name");
				} else if (asked.equals("concat")) {
					operation = new Concat(Arrays.asList("name", "dept"));
				} else if (asked.equals("sort")) {
					operation = new Sort("name");
				}
				if (operation != null) {
					long start = System.currentTimeMillis();
					people = operation.apply(people);
					List<String> collect = people.collect();
					System.out.println("In "
							+ (System.currentTimeMillis() - start) + " ms");
					System.out.println(collect);

				}
			}

			System.out.print(" ? ");
			asked = br.readLine();
		}
	}

	private static String rowToString(Row row) {
		StringBuilder sb = new StringBuilder();
		for (int i = 0; i < row.length(); i++) {
			sb.append(row.get(i) + "/");
		}
		return sb.toString();
	}
}

@@ -834,20 +834,66 @@
 les/
-people.json%22
+customers_10k.json%22);%0A%0A%09%09System.out.println(people.count()
 );%0A%09
@@ -1618,24 +1618,29 @@
 erCase(%22
+first
 name%22);%0A
 %09%09%09%09%7D el
@@ -1627,24 +1627,24 @@
 irstname%22);%0A
-
 %09%09%09%09%7D else i
@@ -1715,28 +1715,44 @@
 asList(%22
+first
 name%22,
- %22dept
+%0A%09%09%09%09%09%09%09%22lastname
 %22));%0A%09%09%09
@@ -1814,20 +1814,20 @@
 w Sort(%22
-name
+city
 %22);%0A%09%09%09%09
@@ -2105,16 +2105,31 @@
 (collect
+.subList(0, 10)
 );%0A%0A%09%09%09%09

use bigger json file