code: 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");
  		
  	}
docs: *************************************************************************