基於Scala的產品開發實踐

我們的產品架構

整體架構

我們的產品代號為Mort（這個代號來自電影《馬達加斯加》那隻萌萌的大眼猴），是基於大數據平台的商業智能（BI）產品。產品架構如下所示：

我們選擇了Spark作為我們的大數據分析平台。基於目前的應用場景，主要使用了Spark SQL，目前使用的版本為Spark 1.5.0。我們有計划去同步升級Spark最新版本。

在研發期間，我們從Spark 1.4升級到1.5，經過性能測評的Benchmark，性能確有顯著提高。Spark 1.6版本在內存管理方面有明顯的改善，Execution Memory與Store Memory的比例可以動態分配，但經過測試，產品的主要性能瓶頸其實是CPU，因為產品的數據分析功能屬於計算密集型。這是我們暫時沒有考慮升級1.6的主因。

從第一次升級Spark的性能測評，以及我們對這一年來Spark版本演進的觀察，我們對Spark的未來充滿信心，尤其是Tungsten項目計劃，會在內存管理、代碼生成以及緩存管理等多方面都會有所提高，對於我們產品而言，算是「坐享其成」了。

由於我們要分析的維度和指標是由客戶指定的，這就需要數據分析的聚合操作是靈活可定製的。因此，我們的產品寫了一個簡單的語法Parser，用以組裝Spark SQL的SQL語句，用以執行分析，最後將DataFrame轉換為我們期待的數據結構返回給前端。

但是，這種設計方案其實牽涉到兩層解析的性能損耗，一個是我們自己的語法Parser，另一個是Spark SQL提供的Parser（通過它將其解析為DataFrame的API調用）。我們考慮在將來會調整方案，直接將客戶定製的聚合操作解析為對DataFrame的API調用（可能會使用新版本Spark的DataSet）。

微服務架構

我們的產品需要支持多種數據源，對數據源的訪問是由另外一個standalone的服務CData完成的，通過它可以隔離這種數據源的多樣性。這相當於一個簡單的微服務架構，目前僅提供兩個服務，一個服務用於數據分析，一個服務用於對客戶數據源的處理：

未來，我們的產品不止限於現有的兩個服務，例如我正在考慮將定期的郵件導出服務獨立出來，保證該服務的獨立性，避免受到其他功能執行的影響。因為這個功能一旦失敗，可能會對客戶的業務產生重要影響。

然而，我們還是在理智地控制服務的粒度。我們不希望因為盲目地追求微服務架構，而帶來運維上的成本。

元數據架構

我們的產品需要存儲元數據（Metadata），用以支持Report、Dashboard以及數據分析，主要的數據模型結果如圖所示：

針對元數據的處理邏輯，我們將之分為職責清晰的三層架構。自上而下分別為REST路由層、應用服務層和元數據資源庫層。

• REST路由層：將元數據視為資源，響應客戶端的HTTP請求，並利用Spray Route將請求路由到對應的動詞上。路由層為核心資源提供Router的trait。這些Router只負責處理客戶端請求，以及服務端的響應，不應包含具體的業務邏輯。傳遞的消息格式為Json格式，由Spray實現消息到Json數據的序列化與反序列化。

• 應用服務層：每個應用服務對應元數據資源的操作用例。由於Mort對元數據的操作並沒有非常複雜的業務邏輯，因此這些服務實際上成為了Router到Repository的中轉站，目的是為了隔離REST路由層對元數據資源庫的依賴。每個服務都被細分為Creator、Editor、Fetcher與Destroyer這樣四個細粒度的trait，並放在對應服務的同一個scala文件中。同時，應用服務要負責保障元數據操作的數據完整性和一致性，因而引入了橫切關注點（Cross Concern Points）中的事務管理。同時，對操作的驗證以及許可權和授權操作也會放到應用服務中。

• 元數據資源庫層：每個資源庫對象都是一個Scala Object，並對應著資料庫中的元數據表。這些對象中的CRUD操作都是原子操作。事實上我們可以認為每個資源庫對象就是元數據的訪問入口。在其實現中，實際上封裝了scalikejdbc的訪問邏輯。

REST路由層和應用服務層需要接收和返回的消息非常相似，甚至在某些場景中，消息結構完全相同，但我們仍然定義了兩套消息體系（皆被定義為Case Class）。邏輯層與消息之間的關係如下圖所示：

在REST路由層，所有的消息皆以Request或Response作為類的後綴名，並被定義為Scala的Case Class。在應用服務層以及元數據資源庫層使用的消息對象則被單獨定義在Messages模塊中。此外，元數據資源庫層還會訪問由ScalikeJDBC生成的Model對象。

我們的技術選型

開發語言的選型

我們選擇的語言是Scala。選擇它的一個主因是因為Spark；另一個原因呢？或許是因為我確實不想再寫Java代碼了。

其實有時候我覺得語言的選型是沒有什麼道理的。除了特殊的應用場景，幾乎所有的程序設計語言都能滿足如今的軟體開發需求。所以我悲哀地看到，語言的紛爭成了宗教的紛爭。

在我們團隊，有熟悉Java的、有熟悉JavaScript包括NodeJS的，有熟悉Clojure的，當然也有熟悉Scala的。除了NodeJS，後端開發幾乎都在JVM平台下。

我對語言選型的判斷標準是：實用、高效、簡潔、可維護。我對Java沒有成見，但我始終認為：即使引入了Lambda以及Method Reference，Java 8在語法方面還是太冗長了。

Scala似乎從誕生開始，一直爭議很大。早在2014年1月ThoughtWorks的Tech Radar中，就講Scala列入了Adopt圈中，但卻在其中特別標註了「the good parts」：

在Top Paying Tech調查中，在美國學習Spark和Scala所值不菲，居然並列冠軍：

其實有了微服務，在不影響代碼維護性的情況下，使用多語言進行開發也成為了可能。或許在將來，我們產品的可能會用clojure或者Ruby來寫DSL，用NodeJS負責元數據（以避免Spray + JSON4S不太好的Json對象序列化）。

說明：將元數據管理單獨獨立為一個NodeJS服務，已經列到了後續架構演進的計劃中。針對元數據管理，我們會統一成JavaScript技術棧，從前端到後端再到資料庫，統一為React+ES6、NodeJS和MongoDB。

坦白說，我沒有強烈的語言傾向性。

數據集的選型

我們還有一個最初的技術選型，後來被認為是失敗的選擇。

CData服務需要將客戶的數據源經過簡單的ETL導入到系統中，我們稱之為數據集（DataSet）。最初在進行技術選型時，我先後考慮過MySQL、Cassandra、HBase。後面兩種都屬於列式存儲的NoSQL資料庫。團隊中沒有一個人有Cassandra的經驗，至於HBase，雖然支持高效的數據查詢，但對聚合運算的支持明顯不足，不適合我們的場景。再加上團隊中有一位成員比較熟悉MySQL，我最終決定使用MySQL。

然而，我們的產品需要支持大數據，當數據量上升到一定級別時，就需要系統很好地支持水平擴展，通過增加更多機器來滿足性能上的需求。評估我們的架構，後端平台可以簡單劃分為三個層次：Web應用服務層（Spray + Nginix）、數據分析層（MESOS + Spark）以及存儲層（主要用於存儲分析數據DataSet，MySQL）。顯然，MySQL會成為水平伸縮的最大障礙。

還好我們醒悟得早，在項目初期就否定了這個方案，而改為採用HDFS+Parquet。

Parquet文件是一種列式數據存儲結構，對於主要為分析型查詢方式的BI數據操作，能夠提供更好的查詢性能。同時，Parquet文件存儲的內容以二進位形式存放，相較於文本形式容量更小，可以節省更多的存儲空間。

Spark SQL提供了對訪問Parquet文件很好的集成。將Parquet文件存放到HDFS中，然後再通過Spark SQL訪問，可以保證在存儲層與數據分析層都能很好地支持分散式處理，從而保證系統的水平伸縮。當對大規模數據集進行分析處理時，可以通過水平增加更多的節點來滿足高性能的實時查詢要求。

我們曾經比較了Parquet方案與MySQL方案，在同等配置下前者的性能要遠遠優於後者，且Spark對Parquet的支持也要好於MySQL。

為了更好地提升性能，我們還計劃在HDFS層之上引入Tachyon，充分發揮內存的優勢，減少磁碟IO帶來的性能損耗。

前端的技術選型

前端的技術選型則為React + Redux。選擇React的原因很簡單，一方面我們認為這種component方式的前端開發，可以極大地提高UI控制項的重用，另一方面，我們認為React這種虛擬DOM的方式在性能上存在一定優勢。此外，React的學習曲線也不高，很容易上手。我們招了3個大學還未畢業的實習生，JS基礎非常薄弱，在我們的培養下，一周後就可以慢慢開始完成React Component開發的小Story了。

在最初的團隊，我們僅有一位前端開發。他選擇了使用CoffeeScript來開發React，但是在項目早期，我們還是忍痛去掉了這些代碼，改為使用ES 6。畢竟隨著ES 6乃至ES 7的普及，JS的標準已經變得越來越合理，CoffeeScript的生存空間似乎被壓縮了。

在前端技術選型方面，我們經歷了好幾次演變。從CoffeeScript到ES 6，從Reflux到Redux，每次變化都在一定程度上增加了工作量。我在文章《技術選型的理想與現實》中講述的就是這個故事。

在《技術選型的理想與現實》這篇文章中，我講到我們選擇了Reflux。然而到現在，最終還是遷移到了Redux。我們一開始並沒有用好Redux，最近的一次重構才讓代碼更符合Redux的最佳實踐。

結論

技術負責人一個非常重要的能力要求就是——善於做出好的技術決策。選擇技術時，並不能一味追求新技術，也不能以自我為中心，選擇「我」認為好的技術。而應該根據產品的需求場景、可能的技術風險、團隊成員能力，並通過分析未來的技術發展趨勢綜合地判斷。

技術決策不可能一成不變，需要與時俱進。如果發現決策錯誤，應該及時糾正，不要遲疑，更不要擔心會影響自己的技術聲譽。

我們的技術實踐

與大多數團隊相比，因為我們使用了小眾的Scala，可以算得上是「撈偏門」了，所以總結的技術實踐未必具有普適性，但對於同為Scala的友朋，或許值得借鑒一二。Scala社區發出的聲音還是太小，有點孤獨——「鸚其鳴也，求其友聲」。

這些實踐不是書本上的創作，而是在產品研發中逐漸演化而來，甚至一些實踐會非常細節。不過，那個優秀的產品不是靠這些細節堆砌出來的呢？

Scala語言的技術實踐

兩年前我還在ThoughtWorks的時候，與同事楊雲（大魔頭）在一個Scala的大數據項目，利用工作之餘，我結合了一些文檔整理了一份Scala編碼規範，放在了github上：Scala編碼規範與最佳實踐。

我們的產品後端全部由Scala進行開發。對於編寫Scala代碼，我的要求很低，只有兩點：

• 寫出來的代碼儘可能有scala范兒，不要看著像Java代碼

• 不要用Scala中理解太費勁兒的語法，否則不利於維護

對於Scala編程，我們還總結了幾條小原則：

• 將業務盡量分布到小的trait中，然後通過object來組合

• 多用函數或偏函數對邏輯進行抽象

• 用隱式轉換體現關注點分離，既保證了職責的單一性，又保證了API的流暢性

• 用getOrElse來封裝需要兩個分支的模式匹配

• 對於隱式參數或支持類型轉換的隱式調用，應盡量讓import語句離調用近一些；對於增加方法的隱式轉換（相當於C#的擴展方法），則應將import放在文件頭，保持調用代碼的乾淨

• 在一個模塊中，盡量將隱式轉換定義放到implicits命名空間下，除非是特別情況需要放到package object中

• 在不影響可讀性的情況下，且無需封裝任何行為，可以考慮使用tuple，而非case class

• 在合適的地方使用lazy關鍵字

AKKA的技術實踐

我們產品用的AKKA並不夠深入，僅僅使用了AKKA的基本功能。主要用於處理前端發來的數據分析消息，相當於一個dispatcher，也承擔了部分消息處理的職責，例如對消息包含的元數據進行解析，生成SQL語句，用以發送給Spark的SqlContext。分析的結果則以Future的方式返回給Spray。

幾條AKKA實踐的小原則：

• actor接收的消息可以分為command和event兩類。命名時，前者用動賓短語，表現為命令請求；後者則使用過去時態，體現fact的本質。

• 產品需要支持多種數據源，不同數據源的處理邏輯放到不同的模塊中，我們利用actor來解耦

以下是為AKKA的ActorRefFactory定義的工廠方法：

trait ActorSupport {nnimplicit val requestTimeout: Timeout = ActorConfig.requestTimeoutnn def actorOf(className: String)(implicit refFactory: ActorRefFactory, trackID: TrackID = random): ActorRef =n refFactory.actorOf(new Props(Props.defaultDeploy, Class.forName(className).asInstanceOf[Class[Actor]], List.empty), id(className))nndef actorOf[T <: Actor : ClassTag](implicit refFactory: ActorRefFactory, trackID: TrackID = random): ActorRef =n refFactory.actorOf(Props[T], id(classTag[T].toString))nndef actorOf[T <: Actor : ClassTag](initial: ActorRefFactory)(implicit trackID: TrackID = random): ActorRef =n initial.actorOf(Props[T], id(classTag[T].toString))n}n

通過向自定義的工廠方法actorOf()傳入Actor的名稱來創建Actor：

def importDataSetData(dataSetId: ID) {nval importDataSetDataActor = actorOf(actorByPersistence("import"))(actorRefFactory)n importDataSetDataActor ! ImportDataSet(dataSetId)n}nndef createDataSetPersistence: Future[Any] = {nval createDataSetPersistenceActor = actorOf(actorByPersistence("create"))(actorRefFactory)n createDataSetPersistenceActor ? dataSetn}n

• 注意actor的sender不能離開當前的ActorContext

• 採用類似Template Method模式的方式去擴展Actor

trait ActorExceptionHandler extends MortActor {n self: Actor =>nnoverride def receive: Receive = {ncase any: Any =>ntry {nsuper.receive(any)n } catch {ncase notFound: ActorNotFound =>nval errorMsg: String = s"invalid parameters: ${notFound.toString}"n log.error(errorMsg)n exceptionSender ! ExecutionFailed(BadRequestException(s"invalid parameters ${notFound.getMessage}"), errorMsg)ncase e: Throwable => exceptionSender ! ExecutionFailed(withTrackID(e, context.self.path.toString), e.getMessage)n }n }nndef exceptionSender = sendern}n

• 或者以類似Decorator模式擴展Actor

trait DelegationActor extends MortActor {nnthis: Actor =>nnprivate val executionResultHandler: Receive = {ncase _: ExecutionResult =>n }nnoverride def receive: Receive = {ncase any: Any =>ntry {n (mortReceive orElse executionResultHandler) (any)n } catch {ncase e: Throwable =>n log.error(e, "")nself ! ExecutionFailed(e)nthrow en } finally {n any match {ncase _: ExecutionResult => self ! PoisonPillncase _ =>n }n }n }n}n

• 考慮建立符合項目要求的SupervisorStrategy

• 盡量利用actor之間的協作來傳遞消息，這樣就可以盡量使用tell而不是ask

Spark SQL的技術實踐

目前的產品特性還未用到更高級的Spark功能。針對一些特殊的客戶，我們計劃採用Spark Streaming來進行流處理，除此之外，核心的數據分析功能都是使用Spark SQL。

以下是我們的一些總結：

• 要學會使用Spark Web UI來幫助我們分析運行指標；另外，Spark本身提供了與Monitoring有關的REST介面，可以集成到自己的系統中；

• 考慮在集群環境下使用Kryo serialization；

• 讓參與運算的數據與運算儘可能地近，在SparkConf中注意設置spark.locality值。注意，需要在不同的部署環境下修改不同的locality值；

• 考慮Spark SQL與性能有關的配置項，例如spark.sql.inMemoryColumnarStorage.batchSize和spark.sql.shuffle.partitions；

• Spark SQL自身對SQL執行定義了執行計劃，而且從執行結果來看，對SQL執行的中間結果進行了緩存，提高了執行的性能。例如我針對相同量級的數據在相同環境下，連續執行了如下三條SQL語句：

第一次執行的SQL語句：

SELECT UniqueCarrier,Origin,count(distinct(Year)) AS Year FROM airline GROUP BY UniqueCarrier,Originn

第二次執行的SQL語句：

SELECT UniqueCarrier,Dest,count(distinct(Year)) AS Year FROM airline GROUP BY UniqueCarrier,Destn

第三次執行的SQL語句：

SELECT Dest , Origin , count(distinct(Year)) AS Year FROM airline GROUP BY Dest , Originn

觀察執行的結果如下所示：

觀察執行count操作的job，顯然第一次執行SQL時的耗時最長，達到2s，而另外兩個job執行的時間則不到一秒。

• 針對複雜的數據分析，要學會充分利用Spark提供的函數擴展機制：UDF（（User Defined Function）與UDAF（User Defined Aggregation Function）；詳細內容，請閱讀文章《Spark強大的函數擴展功能》。

React+Redux的技術實踐

我們一開始並沒有用好React+Redux。隨著對它們的逐漸熟悉，結合社區的一些實踐，我們慢慢體會到了其中的一些好處，也摸索出一些好的實踐。

• 遵循組件設計的原則，我們將React組件分為Component與Container兩種，前者為純組件。

組件設計的原則

• 一個純組件利用props接受所有它需要的數據，類似一個函數的入參，除此之外它不會被任何其它因素影響；

• 一個純組件通常沒有內部狀態。它用來渲染的數據完全來自於輸入props，使用相同的props來渲染相同的純組件多次，

• 將得到相同的UI。不存在隱藏的內部狀態導致渲染不同。

• 在React中儘可能使用extends而不是mixin；

• 對State進行範式化，不要定義嵌套的State結構，不同數據的相互引用都通過ID來查找。範式化的state可以更有效地利用Store里存儲空間；

• 如果不能更改後端返回的模型，可以考慮使用normalizr；但在我們的項目中，為了滿足這一要求，我們專門修改了後端的API。因為採用了之前介紹的元數據架構，這個修改主要影響到了REST路由層和應用服務層的部分代碼；

• 遵循Redux的三大基本原則；

Redux的三大基本原則

• 單一數據源

• State 是只讀的

• 使用純函數來執行修改

在我們的項目中，將所有向後台發送非同步請求的操作都封裝到service中，action會調用這些服務。我們使用了redux-actions的createAction創建dispatch需要的消息：

export const loadDataSource = (id) => {nreturn dispatch => {nreturn DataSourceServices.getDataSource(id)n .then(dataSource => {n dispatch(createAction(DataSourceActionTypes.DATA_SOURCE_RECEIVED)(dataSource))n })n }n}n

在Reducer中，通過redux-actions的handleAction來處理action，避免使用醜陋的switch語句：

export const dataSources = handleActions({n [DataSourceActionTypes.DATA_SOURCES_RECEIVED]: (state, {payload}) => {nconst newState = reduce(payload, (result, dataSource) => {nset(result, dataSource.id, dataSource)nreturn resultn }, state)nreturn assign({}, newState)n },n [DataSourceActionTypes.DATA_SOURCE_RECEIVED]: (state, {payload}) => {nset(state, payload.id, payload)nreturn assign({}, state)n },n [DataSourceActionTypes.DATA_SOURCE_DELETED]: (state, {payload}) => {nreturn omit(state, payload)n }n}, {})n

在Container組件中，如果Store裡面的模型對象需要根據id進行filter或merge之類的操作，則交給selector對其進行封裝。於是Container組件中就可以這樣來調用：

@connect(state => {nreturn {ndataSourcesOfDirectory: DataSourcesSelectors.getDataSourcesOfDirectory(state),n dataSetsOfDataSource: DataSetsSelectors.getDataSetsOfDataSource(state),n selectedDataSource: DataSourcesSelectors.getSelectedDataSource(state),n currentDirectory: DataSourcesSelectors.getCurrentDirectory(state),n memories: state.next.commons.memoriesn }n}, {nloadDataSourcesOfDirectory: DataSourcesActions.loadDataSourcesOfDirectory,n selectDataSource: selectedDataSourceAction.selectDataSource,n cleanSelectedDataSource: selectedDataSourceAction.cleanSelectedDataSource,n loadDataSetsOfDataSource: DataSetsActions.loadDataSetsOfDataSource,n updateDataSource: DataSourcesActions.updateDataSource,n deleteDataSource: DataSourcesActions.deleteDataSource,n navigate: commonActions.navigate,n memory: memoryActions.memory,n cleanMemory: memoryActions.cleanMemory,n goToNewDataSource: NavigationActions.goToNewDataSourcen})n

• 使用eslint來檢查代碼是否遵循ES編寫規範；為了避免團隊成員編寫的代碼不遵守這個規範，甚至可以在git push之前將lint檢查加入到hook中：

echo "npm run lint" > .git/hooks/pre-pushnchmod +x .git/hooks/pre-pushn

Spray與REST的技術實踐

我們的一些總結：

• 站在資源（名詞）的角度去思考REST服務，並遵循REST的規範；

• 考慮GET、PUT、POST、DELETE的安全性與冪等性；

• 必須為REST服務編寫API文檔，並及時更新；

• 使用REST CLIENT對REST服務進行測試，而不能盲目地信任Spray提供的ScalatestRouteTest對客戶端請求的模擬，因為這種模擬其實省略了對Json對象的序列化與反序列化；

• 為核心的REST服務提供健康服務檢查；

• 在Spray中，盡量將自定義的HttpService定義為trait，這樣更利於對它的測試；在自定義的HttpService中，採用cake pattern（使用Self Type）的方式將HttpService注入；

• 我個人不太喜歡Spray以DSL方式編寫REST服務，因為它可能讓函數的嵌套層次太深；如果在一個HttpService（在我們的項目中，皆命名為Router）中，提供的服務較多，建議將各個REST動作都抽取為一個返回Route對象的私有函數，然後利用RouteConcatenation的~運算符拼接起來，以便於閱讀：

def reportRoute(implicit userId: ID) = pathPrefix("reports") {n getReport ~n getViewsOfReport ~n createReport ~n updateReport ~n deleteReport ~n getVirtualField ~n getVirtualFields ~n fuzzyMatch ~n createVirtualFieldn}n

• Spray默認對Json序列化的支持是使用的是Json4s，為此Spray提供了Json4sSupport trait；如果需要支持更多自定義類型的Json序列化，需要重寫隱式值json4sFormats；建議將這些隱式定義放到Object中，交由Router引用，而不是定義為trait去繼承。因為並非Router都使用Json格式，由於trait定義的繼承傳遞性，可能會導致未使用Json格式的Router出現錯誤；

• Json4s可以支持Scala的大多數類型，包括Option等，但不能很好地支持Scala枚舉以及複雜的嵌套遞歸結構，包括多態。這時需要自定義Serializer。具體細節請閱讀我的文章《Spray中對複雜Json的序列化與反序列化》。

推薦閱讀：