干貨 | 智能網聯汽車大數據基礎平臺構建研究

各位老師大家上午好，我們組的題目是智能網聯汽車大數據基礎平臺的構建。我們的指導企業是西部智聯。我們的匯報將從這五個方面進行展開，第一個方面是項目背景與需求分析。

智能網聯汽車是推進智慧交通發展，實現智慧城市建設的重要一環，也是國際公認的未來發展方向，也是我們中國能夠實現汽車工業轉型、形成國際競爭力的一個重要機遇。因此，重慶市政府西部科學城、重慶高新區管委會和我們學校的李克強教授團隊共同策劃成立了智能網聯汽車領域的研發機構——西部科學城智能網聯汽車創新中心。西部智聯也是廣泛開展業務，推進產業協同發展，促進智能網聯汽車產業的規模化落地。

智能網聯汽車云控平臺的用戶主要是普通車主、政府以及企業用戶三個部分。普通車主的需求主要是解決應用協同的問題，保證體驗感。對于政府的話，主要是要統籌整個交通的管理與控制，對于企業用戶的話是利用數據賦能推進產業研發的瓶頸。因此我們就需要不斷建立健全車端云、邊緣云、區域云、全國云四級的支撐體系。

對于市場前景的話，智能交通市場未來的發展也在持續增長，整個市場規模也是非常巨大的，車路云一體化是實現整個智慧交通商業化落地的一個最可行的方案。因此我國也正在大力推動智能網聯汽車示范區的建設。提出到2025年，智能網聯汽車將會覆蓋全國的所有城市。

小米還有集度汽車也采用Flink和Kafka進行了一些數據集成的實踐，我們也建立了北京亦莊自動駕駛示范區以及國家智能網聯汽車長沙測試區，實現了智慧公交、自動駕駛出租車等應用。

接下來我來介紹一下整個系統的設計架構。首先，在數據傳輸階段，我們采用了MQTT協議進行消息的發布，還有訂閱提供一對多的消息發布。在數據處理分發的方面，我們首先采用了Apache Nifi用于可視化的實時整理數據流動，整合數據的總線，接著運用了Kafka對數據進行分發，分發給不同的消費者。在數據分析方面，我們采用了Apache Flink進行有狀態的計算，接著把數據分析后的數據傳入一個TimescaleDB里面。TimescaleDB是基于postgreSQL數據庫打造的一個時序數據庫。數據可以自動按照時間和空間進行一個分區。在數據應用方面，我們運用了Apache的Superset進行可視化，因為它可以支持多種的數據庫，包括連接我們之前用到的TimescaleDB數據庫。

我們整體的系統的架構如上圖所示。除了上述講到的一些技術以外，在第一部分的數據接入，我們采用的數據是企業提供的一個RCU路端數據，還有一個OBU的云車端的數據，然后采用了TCP協議進行通信，然后運用EMQX的引擎進行數據的接入。基于這個系統框架，我們進行后續的開發實踐。

下面我來介紹一下技術實現的部分。首先是我們開發工作的基礎，我們開發工作是基于三臺華為云的Linux機器，這里非常感謝企業導師的支持。然后初始機的環境的配置、開發用的數據主要是導師提供的靜態數據，一共是40萬條，還有數據源的模擬程序，所以最終的動態數據的量級可以達到千萬的量級。導師也提供了一個數據交互規范，供我們進行參考。

數據源程序是使用GO語言進行開發的，我們是使用的MQTT的TCP協議進行傳輸，模擬實際的數據。單個程序的數據源是10赫茲，我們發送的是一個序列化的Json數據，但這個程序可以啟動多個進程，可以模擬多車多路側單元。我們設計的QoS是0，最多分發一次，這樣網絡負載比較低，雖然數據有可能會丟失，但是對我們整個實驗沒有什么太大的影響，下面是一個這個數據的示例。

然后是EMQX的部署，這個部署就比較簡單。需要說明的是EMQX安裝在服務器上，所以需要考慮到網絡安全的問題，所以我們最終還是沒有對公網進行開放，我們把端口轉發到本地進行調試。右上角就是我們安裝完之后的dashboard，我們可以在里面很輕松地看到每秒傳入和傳出的信息量。
接下來是NiFi和Kafka兩個部分，Kafka需要特別注意一下，因為我們后續是在不同的機器上進行流式計算，所以我們需要在局域網內其他機器去訪問Kafka，所以我們還需要修改一下相關的監聽地址。然后就是NiFi的配置，這個配置主要是兩個方面，一個是連接EMQX，是一個MQTT的消費者，然后是連接Kafka，它是一個Kafka brokers，是一個生產者。我們通過同樣的方式訂閱了OBU所有的信息，把數據直接傳輸到Kafka中。

關于Flink的部署，我們在服務器端主要也是考慮到負載平衡和數據吞吐的一些原因，把之前網絡密集型的放在了服務器一上，現在這個計算密集型的我們就安裝在了服務器二。導師推薦的java環境是JDK1.8，所以我們就用了之前的Flink 1.14.6版本。在本地也需要同樣配置一個java開發環境，maven的一個倉庫，添加一個相關的價包，然后就可以用這個Flink的流處理模板去進行開發了。右下角是我們做的Kafka source示例，主要是添加一下相關的地址信息和訂閱的主題，就可以獲得相關的一些offset，配置一下反序列化的方法就可以了。

之后是TimescaleDB的連接，主要是我們需要設計一個存儲表，我們是用的JDBC Sink，然后把Flink的計算結果直接存到TimescaleDB中。后面我們驗證了一下這個數據庫連接是否成功，我們在postgres admin中直接查看數據的結果，可以看到它已經成功存入到數據中了。

然后是Superset和它的連接，我們是用的這種容器化的部署方式，就是用docker去部署，但是這個其實有一些問題，因為容易出現網絡訪問的問題，它需要訪問速手機局域網內的一個其他的服務器，所以還是要去加一些網關方面的操作。在這個配置完成之后，可以看到這個最大的這張圖，實現了數據的可視化。

接下來我們簡單看一些成果展示。首先，最大的成果就是我們實現了從原始數據，也就是從最開始的RCU、OBU的數據到最后呈現到用戶手里的Superset,我們可以用可視化的方法實際看到或者探索數據。下面是服務器的部署，為了平和三臺服務器上的工作，讓負載相對均衡一點，我們把三臺服務器上安裝的分別的東西都列舉在這兒了。第二個部分就是我們的創新點，能夠處理實時流數據。從一開始RCU、OBU到EMQX，我們對于流數據的支持進行了一定的考慮，一直到最后，整個過程都是支持完全支持流數據處理的。第三就是整個數據和架構都具備橫向拓展的空間，我們現在測試用的原始數據是40萬條，但是一方面這個數據庫我們可以拓展，比如說sink到timescaleDB甚至更多的數據庫。最后的可視化也不是只有在Superset，可以在從數據庫中取數據到任何其他你想要的地方，包括從前面我們NiFi、Kafka對于這個topic的管理都是非常完善的，我們有更多的topic，可以有更多的生產者，消費者。最后就是我們在做的這個過程中間也把這些軟件程序的應用給服務化了，服務化的好處就是能夠保證這些服務都能在后臺去運行。我們借助在這個特點，再借助我們用的圖形界面，我們不需要在登錄服務器做什么操作了，只需要在本地端在命令行里邊輸入一段命令，就可以直接在本地的localhost中看到每一個部分的可視化界面，方便我們后續的操作和測試。

之后就是Flink的計算。Flink除了對消息的轉發之外，也包括計算能力。我們這里通過一個簡單的例子去測試一下Flink的計算。這里我們對于輸入進來的數據分類統計每一個時間段的數量，我們在Flink里添加了一些相應的計算代碼之后，把數據也sink到了Times DB的數據庫當中。在這個TimesscaleDB數據當中，我們就可以看到由Flink計算得到的數據。它在TimesscaleDB中呈現出一個數據積累的效果，那么就驗證了在Flink有狀態計算的功能。

另外就是Superset的數據可視化，我們做了一個初步的示例dashboard。可以看出來，右上角的兩個圖對于來源數據進行分類統計，然后左邊這個圖是對于所有車的數據——速度進行了實時的分布統計圖。右下角是其中一輛車，比如說是它的代號是mk007，這輛車它的軌跡樣式，以及所有車的所有數據點的heat map。這里heat map我們可以看出就是在哪些地方，它明顯是一個路口。這里數據點是比較多的，有可能是因為車在這速度比較慢，或者說車在這甚至停掉了。那么在這個heat map可以看到所有數據點的分布情況。
最后簡單總結和反思一下，我們在整個項目過程中間定期的組成例會，每次例會都留有議題文件以及文件內的詳情，每次會議我們都有企業導師去參與進行指導。最后我們整個團隊的合作就依照著技術路線來展開，一步一步來做，步步為營，統籌技術，文字等等去發揮每一位同學的長處優勢。
關于收獲，整個過程大家都收獲了很多，更多的是從技術層面對于大數據相關的技術以及一些更具體的技術都有了更深的了解，當然這個過程中間還有一些存在的問題。最后感謝所有人的努力。