基于Scala的產品開發實踐

ohew2481 7年前發布 | 29K 次閱讀 Scala 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”：

在2016年Stack Overflow發布的開發人員調查結果中，我們也收獲了一些信心。在最愛語言的調查中，Scala排在了第四名：

在引領技術趨勢的調查中，我們選用的React與Spark分列冠亞軍：

在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 {
  implicit val requestTimeout: Timeout = ActorConfig.requestTimeout
def actorOf(className: String)(implicit refFactory: ActorRefFactory, trackID: TrackID = random): ActorRef = refFactory.actorOf(new Props(Props.defaultDeploy, Class.forName(className).asInstanceOf[Class[Actor]], List.empty), id(className))
  def actorOfT <: Actor : ClassTag: ActorRef = refFactory.actorOf(Props[T], id(classTag[T].toString))
  def actorOfT <: Actor : ClassTag(implicit trackID: TrackID = random): ActorRef = initial.actorOf(Props[T], id(classTag[T].toString))
}</code></pre> 
  
通過向自定義的工廠方法actorOf()傳入Actor的名稱來創建Actor： 
  def importDataSetData(dataSetId: ID) {
  val importDataSetDataActor = actorOf(actorByPersistence("import"))(actorRefFactory)

  importDataSetDataActor ! ImportDataSet(dataSetId)
}
def createDataSetPersistence: Future[Any] = {
  val createDataSetPersistenceActor = actorOf(actorByPersistence("create"))(actorRefFactory) 
  createDataSetPersistenceActor ? dataSet
}</code></pre> 
  
 
   注意actor的sender不能離開當前的ActorContext 
   采用類似Template Method模式的方式去擴展Actor 
   
  trait ActorExceptionHandler extends MortActor { 
  self: Actor =>override
def receive: Receive = {
    case any: Any =>
      try {
        super.receive(any) 
      } catch {
        case notFound: ActorNotFound =>
          val errorMsg: String = s"invalid parameters: ${notFound.toString}" 
          log.error(errorMsg) 
          exceptionSender ! ExecutionFailed(BadRequestException(s"invalid parameters ${notFound.getMessage}"), errorMsg)
        case e: Throwable => 
          exceptionSender ! ExecutionFailed(withTrackID(e, context.self.path.toString), e.getMessage) 
      } 
  }
def exceptionSender = sender
}</code></pre> 
  
或者以類似Decorator模式擴展Actor 
  trait DelegationActor extends MortActor {
  this: Actor =>private 
  val executionResultHandler: Receive = {
    case _: ExecutionResult => 
  }
  override def receive: Receive = {
    case any: Any =>
      try { 
        (mortReceive orElse executionResultHandler) (any) 
      } catch {
        case e: Throwable => 
          log.error(e, "")
          self ! ExecutionFailed(e)
          throw e 
      } finally { 
        any match {
          case _: ExecutionResult => self ! PoisonPillcase _ => 
        } 
      } 
  }
} 
   
   考慮建立符合項目要求的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,Origin 
  第二次執行的SQL語句： 
  SELECT UniqueCarrier,Dest,count(distinct(Year)) AS Year FROM airline GROUP BY UniqueCarrier,Dest 
  第三次執行的SQL語句： 
  SELECT Dest , Origin , count(distinct(Year)) AS Year FROM airline GROUP BY Dest , Origin 
  觀察執行的結果如下所示： 
   
  觀察執行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) => {
  return dispatch => {
    return DataSourceServices.getDataSource(id) 
              .then(dataSource => { dispatch(createAction(DataSourceActionTypes.DATA_SOURCE_RECEIVED)(dataSource)) }) 
  }
} 
  在Reducer中，通過redux-actions的handleAction來處理action，避免使用丑陋的switch語句： 
  export const dataSources = handleActions({   
[DataSourceActionTypes.DATA_SOURCES_RECEIVED]: (state, {payload}) => {
    const newState = reduce(payload, (result, dataSource) => {
      set(result, dataSource.id, dataSource)
      return result 
    }, state)
    return assign({}, newState) 
  }, 
[DataSourceActionTypes.DATA_SOURCE_RECEIVED]: (state, {payload}) => {
    set(state, payload.id, payload)
    return assign({}, state) 
  }, 
[DataSourceActionTypes.DATA_SOURCE_DELETED]: (state, {payload}) => {
    return omit(state, payload) }
}, {}) 
  在Container組件中，如果Store里面的模型對象需要根據id進行filter或merge之類的操作，則交給selector對其進行封裝。于是Container組件中就可以這樣來調用： 
  @connect(state => {
  return {
    dataSourcesOfDirectory: DataSourcesSelectors.getDataSourcesOfDirectory(state), 
    dataSetsOfDataSource: DataSetsSelectors.getDataSetsOfDataSource(state), 
    selectedDataSource: DataSourcesSelectors.getSelectedDataSource(state), 
    currentDirectory: DataSourcesSelectors.getCurrentDirectory(state), memories: state.next.commons.memories 
  }
}, {
  loadDataSourcesOfDirectory: DataSourcesActions.loadDataSourcesOfDirectory, 
  selectDataSource: selectedDataSourceAction.selectDataSource, 
  cleanSelectedDataSource: selectedDataSourceAction.cleanSelectedDataSource, 
  loadDataSetsOfDataSource: DataSetsActions.loadDataSetsOfDataSource, 
  updateDataSource: DataSourcesActions.updateDataSource, 
  deleteDataSource: DataSourcesActions.deleteDataSource, 
  navigate: commonActions.navigate, 
  memory: memoryActions.memory, 
  cleanMemory: memoryActions.cleanMemory,   
  goToNewDataSource: NavigationActions.goToNewDataSource
}) 
   
   使用eslint來檢查代碼是否遵循ES編寫規范；為了避免團隊成員編寫的代碼不遵守這個規范，甚至可以在git push之前將lint檢查加入到hook中： 
   
  echo "npm run lint" > .git/hooks/pre-pushchmod +x .git/hooks/pre-push 
  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") {   
  getReport ~ getViewsOfReport ~ createReport ~ updateReport ~ deleteReport ~ getVirtualField ~ getVirtualFields ~ fuzzyMatch ~ createVirtualField
} 
   
   Spray默認對Json序列化的支持是使用的是Json4s，為此Spray提供了Json4sSupport trait；如果需要支持更多自定義類型的Json序列化，需要重寫隱式值json4sFormats；建議將這些隱式定義放到Object中，交由Router引用，而不是定義為trait去繼承。因為并非Router都使用Json格式，由于trait定義的繼承傳遞性，可能會導致未使用Json格式的Router出現錯誤； 
   Json4s可以支持Scala的大多數類型，包括Option等，但不能很好地支持Scala枚舉以及復雜的嵌套遞歸結構，包括多態。這時需要自定義Serializer。 
   
    
  來自：http://zhangyi.farbox.com/post/architecture/our-practice-based-on-scala


                    
                         本文由用戶 ohew2481 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。
                         轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。
                         本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！
                         本文地址：http://www.baiduhome.net/lib/view/open1489461093574.html
                         Scala Scala開發

基于Scala的產品開發實踐

我們的產品架構

我們的技術選型

數據集的選型

我們的技術實踐

React+Redux的技術實踐

相關經驗

相關資訊

相關文檔

目錄