A project for the Distributed Systems course at the Jagiellonian University.

This repo is a fork of poneciak57/hadoop-cluster-docker, which in turn is a fork of kiwenlau/hadoop-cluster-docker.

This project includes a simple traffic simulator and a Map Reduce job running on Apache Hadoop to analize traffic metrics. The solution uses Hadoop Streaming and Map Reduce jobs written in Rust.

A directed graph with checkpoints (nodes) and read segments (edges), represented by a list of checkpoints checkpoints.jsonl and segments segments.jsonl.

Readings from each checkpoint tracking vehicle movements in the form:

{ "checkpoint_id": 123, "vehicle_plate": "KR11FOO12BAR", "time": 21.30005 }

Data is stored in the JSON Lines format.

Inputs can be generated using the binaries generate_network and then simulate_traffic. Use the prepare-trafic-metrics.sh script inside the Docker containers.

The generate_network script generates a network of checkpoints and road segments.

The currently chosen implementation outputs a full binary tree with the depth specified using the --tree_depth flag. This is a terrible approximation of a city street network, but is straightforward to implement, and the resulting street segments have vastly different traffic volumes, generating interesting data to analyze.

The code sources also contain a clique generator. A street network like that is definitely not worth investigating :).

The network is represented by two files:

• checkpoints.jsonl, with metadata used only for readings generation (the number of residents at each checkpoint and the destination weight).
• segments.jsonl, with street names, begin and checkpoint ids, length and speed function.

The simulate_traffic scripts takes the files generated by generate_network and outputs checkpoint readings divided into --input-file-count files in the readings/ directory.

The traffic generation works by generating a Vehicle instance for each resident on a checkpoint. Each Vehicle chooses randomly:

• one workplace (checkpoint) (using Checkpoint.destination_weight as the weight)
• the parameters for a normal distribution for generating the home departure time each day
• the parameters for a normal distribution for generating the time before returning home.

The next step is simulating the movement of a vehicle to its destination. Vehicles choose the next segment which minimizes the distance (in meters) to the destination. The next segments are precomputed for each (source, destination) pair.

The simulator keeps track of the number of cars on each road segment at a given moment. These numbers are used to determine the travel speeds of segments. This is, of course, a very crude approximation.

The readings are split into files by a hash of the segment id. Although the readings in each file are sorted by time, the map-reduce jobs don't rely on this ordering.

• docker compose up --build -d && docker exec -it hadoop-master bash to enter the bash shell, where the comands specified below can be run.
• ./prepare-trafic-metrics.sh --- generates inputs
• ./run-traffic-metrics.sh --- runs all MapReduce jobs sequentially

The Rust code is compiled automatically in the Dockerfile, and all the binaries are made available in PATH. To execute the binary scripts outside of Docker, use:

cd code
cargo run --bin generate-network --tree-depth 5 

cargo will then take care of installing dependencies and compiling the project.

The current set of jobs collects stats about all vehicle trips (a trip is a journey from home to workplace or back):

• The total number of trips
• Sum of the distances for all trips
• Sum of the durations for all trips

These stats get outputted at the stats key in the trip-stats directory. They can be used to get the average trip distance and duration.

The entries with keys longest_distance_trips and longest_duration_trips contain the list of the top 16 (configurable) trips by distance/duration. At most one trip per vehicle plate number is allowed.

flowchart TB
    subgraph job1["Job 1"]
        map1(["map: map_hops"])
        partition1("partition by vehicle plate")
        sort1("sort by time")
        reduce1(["reduce: reduce_hops"])
    end

    subgraph job2["Job 2"]
        map2(["map: map_hops_with_segments"])
        partition2("partition by segment begin and end")
        sort2("sort for each (begin, end) pair: first segment info then hops")
        reduce2(["reduce: reduce_hops_with_segments"])
    end

    subgraph job3["Job 3"]
        map3(["map: map_trips"])
        combine3(["combine: reduce_trips"])
        reduce3(["reduce: reduce_trips"])
    end

    subgraph job4["Job 4"]
        map4(["map: map_trip_stats"])
        combine4(["combine: reduce_trip_stats"])
        reduce4(["reduce: reduce_trip_stats"])
    end

    segments["input/segments.jsonl"]
    readings["input/readings/readings-*.jsonl"]

    hops["hops/"]
    hopsWithSegments["hops-with-segments/"]
    trips["trips/"]
    tripStats["trip-stats/"]

    readings --> map1
    map1 --> partition1
    partition1 --> sort1
    sort1 --> reduce1
    reduce1 --> hops

    hops --> map2
    segments -------> map2
    map2 --> partition2
    partition2 --> sort2
    sort2 --> reduce2
    reduce2 --> hopsWithSegments

    hopsWithSegments --> map3
    map3 --> combine3
    combine3 --> reduce3
    reduce3 --> trips

    trips --> map4
    map4 --> combine4
    combine4 --> reduce4
    reduce4 --> tripStats

• Takes raw input/readings/readings-*.jsonl files for the readings.
• map_hops script maps each reading so that the key has the form <plate>|<time> (KK12314112|123.1512).
• Partition the readings the first key field (<plate>)
• Sort by <plate> first, then by <time> numerically
• reduce_hops maps pairs of consequtive readings for a given vehicle and outputs serialized Hop instances as values (and <from>|<to> as keys).

The output is placed in the hops directory.

• Reads files from the hops directory and the input/segments.jsonl file
• map_hops_with_segments:
  • Maps entries from input/segments.jsonl (serialized Segment instanes) identically, assigning the keys <from>|<to>|1
  • Maps entries from hops identically, assigning the keys <from>|<to>|2
• Keys are partitioned using the first two key fields (<from>|<to>)
• Keys are sorted by <from>, <to> and 1/2, so that the segment info is always first
• reduce_hops_with_segments outputs <from>|<to> as the keys and HopWithSegment as values.

By the end each output entry in the hops_with_segments folder is an entry from hops with the added segment_name and segment_length_m fields.

• Reads files from hops-with-segments
• map_trips maps each hop as a (partial) Trip with <plate>|<vehicle-trip-number> as key
• reduce_trips is used for the combine and reduce steps to merge partial trips into full Trips (from workplace to home or from home to workplace)

The output is placed in trips

• Reads files from trips
• map_trip_stats maps each Trip to three keys:
  • stats with the value beging a TripStats instance containing data only for one Trip
  • longest_distance_trips: A JSON array of 16 trips with the longest distance, sorted decreasingly and with at most one instance per vehicle plate (the map step only outputs arrays of length 1)
  • longest_duration_trips: Same as above but sorted by time, not distance
• reduce_trip_stats is used for the combine and reduce steps:
  • Values with the stats key are summed and outputed with the stats key
  • The values for keys longest_distance_trips and longest_duration_trips are merged (as in merge sort).

The output is placed in trip-stats.

There are more map-reduce jobs, which could be implemented based on the generated data. Here are some other ideas:

• Analyzing traffic on each street by time of day (for example, in 15-minute time slots):
  • Average speed / travel time
  • Number of vehicles
• Analyzing the 15, 50 and 85 percentile speeds on each street.
• Determining the function of travel speed(vehicle count) for each street.
• Identifying peak hours on the network.

Please note that the benchmarks were run on a laptop and CPU throttling got triggered multiple times during these tests.

The fourth stage job started failing during these tests when the number of reducers was set to a value other than 1. I was unable to find the cause for this because of difficulties with retrieving the Rust logs when using Hadoop Streaming. The failing fourth job does not make the results in the rest of the columns invalid.