<commit_before>package de.kdml.bigdatalab.spark_and_flink.flink.datacorn;

import java.util.List;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import de.kdml.bigdatalab.spark_and_flink.common_utils.data.Trajectory;
import de.kdml.bigdatalab.spark_and_flink.flink.utils.FlinkUtils;

/**
 * 
 * @author <NAME>
 * 
 *         Jan 5, 2017
 */
public class TrajectoriesSectorChangeDetector {

	public static void main(String[] args) throws Exception {

		StreamExecutionEnvironment env = FlinkUtils.getInitializedEnv();
		DataStream<Tuple2<String, List<Trajectory>>> trajectoriesStream = TrajectoriesStreamUtils
				.getTrajectoriesStream(env);

		trajectoriesStream.print().setParallelism(1);
		env.execute("Trajectories Sector change detector");
	}
}<commit_msg>Added Flink trajectories stream sector change detector <commit_after>package de.kdml.bigdatalab.spark_and_flink.flink.datacorn;

import java.util.List;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import de.kdml.bigdatalab.spark_and_flink.common_utils.Configs;
import de.kdml.bigdatalab.spark_and_flink.common_utils.data.Sector;
import de.kdml.bigdatalab.spark_and_flink.common_utils.data.Trajectory;
import de.kdml.bigdatalab.spark_and_flink.flink.utils.FlinkUtils;

/**
 * 
 * @author <NAME>
 * 
 *         Jan 5, 2017
 */
public class TrajectoriesSectorChangeDetector {

	private static Configs configs = Configs.getInstance();

	public static void main(String[] args) throws Exception {

		StreamExecutionEnvironment env = FlinkUtils.getInitializedEnv();
		DataStream<Tuple2<String, List<Trajectory>>> trajectoriesStream = TrajectoriesStreamUtils
				.getTrajectoriesStream(env);

		DataSet<Sector> sectorsDataSet = getSectorDataSet();

		// assign sector for each trajectory, then check if there is change in
		// the sector between two consecutive trajectories
		DataStream<String> trajectoriesStreamWithSectors = trajectoriesStream
				.map(new TrajectorySectorMapper(sectorsDataSet.collect())).filter(tuple -> {
					// sort trajectories
					List<Trajectory> trajectories = tuple.f1;
					if (trajectories != null && trajectories.size() > 1) {

						// check for change between trajectories with respect to
						// sectors
						int i = trajectories.size() - 1;
						Trajectory trajectoryi = trajectories.get(i - 1);
						Trajectory trajectoryi1 = trajectories.get(i);

						if ((!trajectoryi.getSector().equals(trajectoryi1.getSector()))) {

							return true;
						}

					}
					return false;
				}).map(tuple -> {

					// print the sector transition
					List<Trajectory> trajectories = tuple.f1;
					int i = trajectories.size() - 1;
					Trajectory trajectoryi = trajectories.get(i - 1);
					Trajectory trajectoryi1 = trajectories.get(i);
					return tuple.f0 + ": " + trajectoryi.getSector().getNameAndAirBlock() + " --> "
							+ trajectoryi1.getSector().getNameAndAirBlock();

				});

		trajectoriesStreamWithSectors.print().setParallelism(1);
		env.execute("Trajectories Sector change detector");
	}

	/***
	 * Get the sectors data set
	 */
	private static DataSet<Sector> getSectorDataSet() {
		// get the sectors list
		ExecutionEnvironment dataSetAPIEnv = ExecutionEnvironment.getExecutionEnvironment();
		DataSet<Sector> sectorsDataSet = dataSetAPIEnv.readTextFile(configs.getStringProp("sectorsFilePath"))
				.setParallelism(1).map(line -> {
					return Sector.parseSectorData(line);
				}).setParallelism(1);
		return sectorsDataSet;
	}
}
{'fraction_non_alphanumeric': 0.11029604401362327, 'fraction_numerical': 0.007859575582918523, 'mean_word_length': 4.501440922190202, 'pattern_counts': {'":': 1, '<': 17, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2377065', 'n_tokens_mistral': 1373, 'n_tokens_neox': 1185, 'n_words': 214}