import java.io.Serializable;
import java.io.FileNotFoundException;
import java.io.File;
import java.util.*;
import java.util.Iterator;
import java.util.List;
import java.util.List;
import org.apache.spark.api.java.*;
import org.apache.spark.api.java.function.Function;
public class SimpleApp implements Serializable{
public static void main(String[] args) {
String logFile = "/tmp/master.txt"; // Should be some file on your system
String positive = "/tmp/positive.txt"; // Should be some file on your system
String negative = "/tmp/negative.txt"; // Should be some file on your system
JavaSparkContext sc = new JavaSparkContext("local[4]", "Twitter Analyzer", "/home/welcome/Downloads/spark-1.1.0/", new String[]{"target/scala-2.10/Simple-assembly-0.1.0.jar"});
JavaRDD<String> positiveComments = sc.textFile(logFile).cache();
List<String> positiveList = GetSentiments(positive);
List<String> negativeList= GetSentiments(negative);
final Iterator<String> iterator = positiveList.iterator();
int i = 0;
while (iterator.hasNext())
{
JavaRDD<String> numAs = positiveComments.filter(new Function<String, Boolean>()
{
public Boolean call(String s)
{
return s.contains(iterator.next());
}
});
numAs.saveAsTextFile("/tmp/output/"+ i);
i++;
}
}
public static List<String> GetSentiments(String fileName) {
List<String> input = new ArrayList<String>();
try
{
Scanner sc = new Scanner(new File(fileName));
while (sc.hasNextLine()) {
input.add(sc.nextLine());
}
}
catch (FileNotFoundException e){
// do stuff here..
}
return input;
}
}