package blah;

import static spark.Spark.get;

import javax.ws.rs.core.Response;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import spark.servlet.SparkApplication;

public class App implements SparkApplication {

	@Override
	public void init() {
		get("/hello", (req, res) -> {
			String sourcePath = "hdfs://spark:54310/input/*";

			SparkConf conf = new SparkConf().setAppName("TestLineCount");
			conf.setJars(new String[] { App.class.getProtectionDomain()
					.getCodeSource().getLocation().getPath() });
			conf.setMaster("spark://tootak:7077");
			conf.set("spark.driver.allowMultipleContexts", "true");

			@SuppressWarnings("resource")
			JavaSparkContext sc = new JavaSparkContext(conf);
			JavaRDD<String> log = sc.textFile(sourcePath);

			JavaRDD<String> lines = log.filter(x -> {
				return true;
			});

			return Response.ok(lines.count()).build();
		});
	}
}