public static void main(String[] args) throws Exception {
		
		if(!parseParameters(args)) {
			return;
		}

		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

		// get input data
		DataSet<Tuple2<String, String>> documents = getDocumentsDataSet(env);
		DataSet<Tuple3<Integer, String, Integer>> ranks = getRanksDataSet(env);
		DataSet<Tuple2<String, String>> visits = getVisitsDataSet(env);
		
		// Retain documents with keywords
		DataSet<Tuple1<String>> filterDocs = documents
				.filter(new FilterDocByKeyWords())
				.project(0).types(String.class);

		// Filter ranks by minimum rank
		DataSet<Tuple3<Integer, String, Integer>> filterRanks = ranks
				.filter(new FilterByRank());

		// Filter visits by visit date
		DataSet<Tuple1<String>> filterVisits = visits
				.filter(new FilterVisitsByDate())
				.project(0).types(String.class);

		// Join the filtered documents and ranks, i.e., get all URLs with min rank and keywords
		DataSet<Tuple3<Integer, String, Integer>> joinDocsRanks = 
				filterDocs.join(filterRanks)
							.where(0).equalTo(1)
							.projectSecond(0,1,2)
							.types(Integer.class, String.class, Integer.class);

		// Anti-join urls with visits, i.e., retain all URLs which have NOT been visited in a certain time
		DataSet<Tuple3<Integer, String, Integer>> result = 
				joinDocsRanks.coGroup(filterVisits)
								.where(1).equalTo(0)
								.with(new AntiJoinVisits());

		// emit result
		if(fileOutput) {
			result.writeAsCsv(outputPath, "\n", "|");
		} else {
			result.print();
		}

		// execute program
		env.execute("WebLogAnalysis Example");
		
	}
public static void main ( String [ ] args ) throws Exception { if ( ! parseParameters ( args ) ) { return ; } final ExecutionEnvironment env = ExecutionEnvironment . getExecutionEnvironment ( ) ; DataSet < Tuple2 < String , String > > documents = getDocumentsDataSet ( env ) ; DataSet < Tuple3 < Integer , String , Integer > > ranks = getRanksDataSet ( env ) ; DataSet < Tuple2 < String , String > > visits = getVisitsDataSet ( env ) ; DataSet < Tuple1 < String > > filterDocs = documents . filter ( new FilterDocByKeyWords ( ) ) . project ( 0 ) . types ( String . class ) ; DataSet < Tuple3 < Integer , String , Integer > > filterRanks = ranks . filter ( new FilterByRank ( ) ) ; DataSet < Tuple1 < String > > filterVisits = visits . filter ( new FilterVisitsByDate ( ) ) . project ( 0 ) . types ( String . class ) ; DataSet < Tuple3 < Integer , String , Integer > > joinDocsRanks = filterDocs . join ( filterRanks ) . where ( 0 ) . equalTo ( 1 ) . projectSecond ( 0 , 1 , 2 ) . types ( Integer . class , String . class , Integer . class ) ; DataSet < Tuple3 < Integer , String , Integer > > result = joinDocsRanks . coGroup ( filterVisits ) . where ( 1 ) . equalTo ( 0 ) . with ( new AntiJoinVisits ( ) ) ; if ( fileOutput ) { result . writeAsCsv ( outputPath , "\n" , "|" ) ; } else { result . print ( ) ; } env . execute ( "WebLogAnalysis Example" ) ; }
*************************************************************************