1. 程式人生 > >Spark專案智慧城市車流量分析專案之固定卡口下車輛的行車軌跡

Spark專案智慧城市車流量分析專案之固定卡口下車輛的行車軌跡

資料集

  • 日期 卡口ID 攝像頭編號 車牌號 拍攝時間
  • date monitor_id camera_id car action_time
    車速 道路ID 區域ID
    speed road_id area_id

資料集

模擬資料

2018-06-27 0007 00536 京R66884 2018-06-27 11:30:25 30 41 08
2018-06-27 0005 01726 閔P89564 2018-06-27 09:34:03 19 7 03
2018-06-27 0005 01272 閔P89564 2018-06-27 09:50:39 187 19 05
2018-06-27 0002 00082 閔P89564 2018-06-27 09:34:47 1 28 05
2018-06-27 0003 08417 閔P89564 2018-06-27 09:23:05 171 42 02
2018-06-27 0003 02757 閔P89564 2018-06-27 09:52:35 32 50 04
2018-06-27 0000 03759 滬W87972 2018-06-27 20:02:43 243 25 05
2018-06-27 0002 08652 滬W87972 2018-06-27 20:05:10 51 11 05

程式碼邏輯
在這裡插入圖片描述

本地模擬程式碼

package test;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import scala.Tuple2;

import java.util.*;

public class Carlearn {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local").setAppName("Carlearn");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> lineRDD = sc.textFile("monitor_flow_action");


        //mapToPair 形成一個tuple<卡扣id,車牌號>
        JavaPairRDD<String, String> rdd2 = lineRDD.mapToPair(new PairFunction<String, String, String>() {
            @Override
            public Tuple2<String, String> call(String s) throws Exception {
                String[] split = s.split("\t");
                return new Tuple2<>(split[1], split[3]);
            }
        });

        //過濾只留下一個卡扣id為001的資訊
        JavaPairRDD<String, String> rdd3 = rdd2.filter(new Function<Tuple2<String, String>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, String> tuple2) throws Exception {

                return tuple2._1.equals("0001");
            }
        });

        //map 形成一個個的car
        JavaRDD<String> rdd4 = rdd3.map(new Function<Tuple2<String, String>, String>() {
            @Override
            public String call(Tuple2<String, String> tuple2) throws Exception {
                return tuple2._2;
            }
        });

        //去重
        JavaRDD<String> rdd5 = rdd4.distinct();

        //轉換成集合
        List<String> result = rdd5.collect();

        //廣播變數
        final Broadcast<List<String>> broadcast = sc.broadcast(result);

        /**
         * mapToPair轉換成Tuple<車牌號,一行的值>
         */
        JavaPairRDD<String, String> rddA = lineRDD.mapToPair(new PairFunction<String, String, String>() {
            @Override
            public Tuple2<String, String> call(String s) throws Exception {
                String[] split = s.split("\t");

                return new Tuple2<>(split[3], s);
            }
        });
        //過濾掉只留下卡扣id為001資訊的車,
        JavaPairRDD<String, Iterable<String>> rddC = rddA.filter(new Function<Tuple2<String, String>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, String> tuple2) throws Exception {
                List<String> carList = broadcast.value();
                String carId = tuple2._2.split("\t")[3];
                return carList.contains(carId);
            }
        }).groupByKey();
        rddC.foreach(new VoidFunction<Tuple2<String, Iterable<String>>>() {
            @Override
            public void call(Tuple2<String, Iterable<String>> tuple2) throws Exception {
                Iterator<String> iterator = tuple2._2.iterator();
                List<Tuple2<String, String>> map = new ArrayList<>();
                //將時間和卡扣號封裝成一個list
                while (iterator.hasNext()) {
                    String next = iterator.next();
                    String time = next.split("\t")[4];
                    String kakouId = next.split("\t")[1];
                    Tuple2<String, String> time_kakou = new Tuple2<>(time, kakouId);
                    map.add(time_kakou);
                }
                //通過時間排序,整理出汽車的行車軌跡
                Collections.sort(map, new Comparator<Tuple2<String, String>>() {
                    @Override
                    public int compare(Tuple2<String, String> o1, Tuple2<String, String> o2) {
                        return o1._1.compareTo(o2._1);
                    }
                });
                //列印下行車軌跡
                String chepai = tuple2._1;
                System.out.println("汽車號:"+chepai);
                for (Tuple2<String, String> s : map) {
                    System.out.print("時間:"+s._1+"卡扣:"+s._2+"===>");
                }
                System.out.println();
            }
        });
    }
}

結果

06-27 22:39:26卡扣:0005===>時間:2018-06-27 22:40:31卡扣:0008===>時間:2018-06-27 22:41:37卡扣:0005===>時間:2018-06-27 22:42:02卡扣:0000===>時間:2018-06-27 22:43:07卡扣:0002===>時間:2018-06-27 22:44:28卡扣:0004===>時間:2018-06-27 22:45:49卡扣:0001===>時間:2018-06-27 22:51:40卡扣:0005===>時間:2018-06-27 22:53:26卡扣:0008===>時間:2018-06-27 22:54:06卡扣:0007===>時間:2018-06-27 22:57:26卡扣:0004===>時間:2018-06-27 22:57:56卡扣:0005===>時間:2018-06-27 22:58:07卡扣:0006===>時間:2018-06-27 23:00:40卡扣:0006===>時間:2018-06-27 23:01:56卡扣:0008===>時間:2018-06-27 23:04:27卡扣:0004===>時間:2018-06-27 23:04:43卡扣:0003===>時間:2018-06-27 23:05:38卡扣:0001===>時間:2018-06-27 23:09:03卡扣:0007===>時間:2018-06-27 23:09:06卡扣:0005===>時間:2018-06-27 23:09:37卡扣:0004===>時間:2018-06-27 23:11:11卡扣:0007===>時間:2018-06-27 23:12:06卡扣:0001===>時間:2018-06-27 23:12:12卡扣:0008===>時間:2018-06-27 23:14:31卡扣:0002===>時間:2018-06-27 23:16:25卡扣:0007===>時間:2018-06-27 23:23:47卡扣:0000===>時間:2018-06-27 23:24:03卡扣:0008===>時間:2018-06-27 23:26:27卡扣:0003===>時間:2018-06-27 23:29:44卡扣:0001===>時間:2018-06-27 23:30:23卡扣:0003===>時間:2018-06-27 23:33:27卡扣:0007===>時間:2018-06-27 23:34:18卡扣:0002===>時間:2018-06-27 23:34:28卡扣:0002===>時間:2018-06-27 23:34:39卡扣:0001===>時間:2018-06-27 23:39:59卡扣:0005===>時間:2018-06-27 23:45:13卡扣:0001===>時間:2018-06-27 23:48:47卡扣:0004===>
汽車號:京D44143
時間:2018-06-27 02:00:01卡扣:0007===>時間:2018-06-27 02:00:59卡扣:0001===>時間:2018-06-27 02:02:11卡扣:0008===>時間:2018-06-27 02:02:29卡扣:0001===>時間:2018-06-27 02:02:31卡扣:0001===>時間:2018-06-27 02:04:11卡扣:0005===>時間:2018-06-27 02:06:20卡扣:0000===>時間:2018-06-27 02:07:36卡扣:0007===>時間:2018-06-27 02:14:36卡扣:0004===>時間:2018-06-27 02:14:58卡扣:0005===>時間:2018-06-27 02:15:01卡扣:0007===>時間:2018-06-27 02:15:14卡扣:0003===>時間:2018-06-27 02:17:16卡扣:0005===>時間:2018-06-27 02:20:59卡扣:0001===>時間:2018-06-27 02:21:26卡扣:0008===>時間:2018-06-27 02:21:27卡扣:0000===>時間:2018-06-27 02:22:19卡扣:0004===>時間:2018-06-27 02:26:20卡扣:0007===>時間:2018-06-27 02:35:02卡扣:0002===>時間:2018-06-27 02:35:28卡扣:0006===>時間:2018-06-27 02:37:23卡扣:0001===>時間:2018-06-27 02:38:13卡扣:0005===>時間:2018-06-27 02:38:33卡扣:0002===>時間:2018-06-27 02:39:15卡扣:0000===>時間:2018-06-27 02:39:42卡扣:0008===>時間:2018-06-27 02:41:53卡扣:0001===>時間:2018-06-27 02:46:54卡扣:0007===>時間:2018-06-27 02:47:23卡扣:0008===>時間:2018-06-27 02:52:27卡扣:0001===>時間:2018-06-27 02:53:31卡扣:0006===>時間:2018-06-27 03:08:55卡扣:0007===>
汽車號:京R34631
時間:2018-06-27 14:01:24卡扣:0004===>時間:2018-06-27 14:03:36卡扣:0005===>時間:2018-06-27 14:05:04卡扣:0006===>時間:2018-06-27 14:08:40卡扣:0007===>時間:2018-06-27 14:09:29卡扣:0002===>時間:2018-06-27 14:09:51卡扣:0008===>時間:2018-06-27 14:13:40卡扣:0008===>時間:2018-06-27 14:18:55卡扣:0005===>時間:2018-06-27 14:26:54卡扣:0001===>時間:2018-06-27 14:27:20卡扣:0000===>時間:2018-06-27 14:34:02卡扣:0001===>時間:2018-06-27 14:34:39卡扣:0006===>時間:2018-06-27 14:36:10卡扣:0003===>時間:2018-06-27 14:37:08卡扣:0006===>時間:2018-06-27 14:37:30卡扣:0003===>時間:2018-06-27 14:39:25卡扣:0005===>時間:2018-06-27 14:41:00卡扣:0007===>時間:2018-06-27 14:42:03卡扣:0001===>時間:2018-06-27 14:47:36卡扣:0001===>時間:2018-06-27 14:48:16卡扣:0003===>時間:2018-06-27 14:53:22卡扣:0001===>時間:2018-06-27 14:53:34卡扣:0005===>時間:2018-06-27 14:53:54卡扣:0001===>時間:2018-06-27 14:53:56卡扣:0002===>時間:2018-06-27 14:54:14卡扣:0003===>時間:2018-06-27 14:55:44卡扣:0003===>時間:2018-06-27 14:57:27卡扣:0006===>時間:2018-06-27 14:57:53卡扣:0000===>時間:2018-06-27 14:58:12卡扣:0008===>時間:2018-06-27 14:59:11卡扣:0008===>
汽車號:滬V82625