Spark Streaming 非常適合ETL。但是其開發(fā)模塊化程度不高，所以這里提供了一套方案，該方案提供了新的API用于開發(fā)Spark Streaming程序，同時(shí)也實(shí)現(xiàn)了模塊化，配置化，并且支持SQL做數(shù)據(jù)處理。

項(xiàng)目地址

前言

傳統(tǒng)的Spark Streaming程序需要：

• 構(gòu)建StreamingContext
• 設(shè)置checkpoint
• 鏈接數(shù)據(jù)源
• 各種transform
• foreachRDD 輸出

通常而言，你可能會(huì)因?yàn)橐咄晟厦娴牧鞒潭鴺?gòu)建了一個(gè)很大的程序，比如一個(gè)main方法里上百行代碼，雖然在開發(fā)小功能上足夠便利，但是復(fù)用度更方面是不夠的，而且不利于協(xié)作，所以需要一個(gè)更高層的開發(fā)包提供支持。

如何開發(fā)一個(gè)Spark Streaming程序

我只要在配置文件添加如下一個(gè)job配置，就可以作為標(biāo)準(zhǔn)的的Spark Streaming 程序提交運(yùn)行：

{

  "test": {
    "desc": "測(cè)試",
    "strategy": "streaming.core.strategy.SparkStreamingStrategy",
    "algorithm": [],
    "ref": [],
    "compositor": [
      {
        "name": "streaming.core.compositor.kafka.MockKafkaStreamingCompositor",
        "params": [
          {
            "metadata.broker.list":"xxx",
            "auto.offset.reset":"largest",
            "topics":"xxx"
          }
        ]
      },
      {
        "name": "streaming.core.compositor.spark.JSONTableCompositor",
        "params": [{"tableName":"test"}
        ]
      },
      {
        "name": "streaming.core.compositor.spark.SQLCompositor",
        "params": [{"sql":"select a from test"}
        ]
      },
      {
        "name": "streaming.core.compositor.RDDPrintOutputCompositor",
        "params": [
          {
          }
        ]
      }
    ],
    "configParams": {
    }
  }
}

上面的配置相當(dāng)于完成了如下的一個(gè)流程：

1. 從Kafka消費(fèi)數(shù)據(jù)
2. 將Kafka數(shù)據(jù)轉(zhuǎn)化為表
3. 通過SQL進(jìn)行處理
4. 打印輸出

是不是很簡(jiǎn)單，而且還可以支持熱加載，動(dòng)態(tài)添加job等

特性

該實(shí)現(xiàn)的特性有：

1. 配置化
2. 支持多Job配置
3. 支持各種數(shù)據(jù)源模塊
4. 支持通過SQL完成數(shù)據(jù)處理
5. 支持多種輸出模塊

未來可擴(kuò)展的支持包含：

1. 動(dòng)態(tài)添加或者刪除job更新，而不用重啟Spark Streaming
2. 支持Storm等其他流式引擎
3. 更好的多job互操作

配置格式說明

該實(shí)現(xiàn)完全基于ServiceframeworkDispatcher 完成，核心功能大概只花了三個(gè)小時(shí)。

這里我們先理出幾個(gè)概念：

1. Spark Streaming 定義為一個(gè)App
2. 每個(gè)Action定義為一個(gè)Job.一個(gè)App可以包含多個(gè)Job

配置文件結(jié)構(gòu)設(shè)計(jì)如下：

{

  "job1": {
    "desc": "測(cè)試",
    "strategy": "streaming.core.strategy.SparkStreamingStrategy",
    "algorithm": [],
    "ref": [],
    "compositor": [
      {
        "name": "streaming.core.compositor.kafka.MockKafkaStreamingCompositor",
        "params": [
          {
            "metadata.broker.list":"xxx",
            "auto.offset.reset":"largest",
            "topics":"xxx"
          }
        ]
      } ,  
    ],
    "configParams": {
    }
  }，
  "job2"：{
   ........
 } 
}

一個(gè)完整的App 對(duì)應(yīng)一個(gè)配置文件。每個(gè)頂層配置選項(xiàng)，如job1,job2分別對(duì)應(yīng)一個(gè)工作流。他們最終都會(huì)運(yùn)行在一個(gè)App上(Spark Streaming實(shí)例上)。

• strategy 用來定義如何組織 compositor,algorithm, ref 的調(diào)用關(guān)系
• algorithm作為數(shù)據(jù)來源
• compositor 數(shù)據(jù)處理鏈路模塊。大部分情況我們都是針對(duì)該接口進(jìn)行開發(fā)
• ref 是對(duì)其他job的引用。通過配合合適的strategy，我們將多個(gè)job組織成一個(gè)新的job
• 每個(gè)組件( compositor,algorithm, strategy) 都支持參數(shù)配置

上面主要是解析了配置文件的形態(tài)，并且ServiceframeworkDispatcher 已經(jīng)給出了一套接口規(guī)范，只要照著實(shí)現(xiàn)就行。

模塊實(shí)現(xiàn)

那對(duì)應(yīng)的模塊是如何實(shí)現(xiàn)的？本質(zhì)是將上面的配置文件，通過已經(jīng)實(shí)現(xiàn)的模塊，轉(zhuǎn)化為Spark Streaming程序。

以SQLCompositor 的具體實(shí)現(xiàn)為例：

class SQLCompositor[T] extends Compositor[T] {

  private var _configParams: util.List[util.Map[Any, Any]] = _
  val logger = Logger.getLogger(classOf[SQLCompositor[T]].getName)

//策略引擎ServiceFrameStrategy 會(huì)調(diào)用該方法將配置傳入進(jìn)來
  override def initialize(typeFilters: util.List[String], configParams: util.List[util.Map[Any, Any]]): Unit = {
    this._configParams = configParams
  }

// 獲取配置的sql語句
  def sql = {
    _configParams(0).get("sql").toString
  }

  def outputTable = {
    _configParams(0).get("outputTable").toString
  }

//執(zhí)行的主方法，大體是從上一個(gè)模塊獲取SQLContext(已經(jīng)注冊(cè)了對(duì)應(yīng)的table),
//然后根據(jù)該模塊的配置，設(shè)置查詢語句，最后得到一個(gè)新的dataFrame.
// middleResult里的T其實(shí)是DStream,我們會(huì)傳遞到下一個(gè)模塊，Output模塊
//params參數(shù)則是方便各個(gè)模塊共享信息，這里我們將對(duì)應(yīng)處理好的函數(shù)傳遞給下一個(gè)模塊
  override def result(alg: util.List[Processor[T]], ref: util.List[Strategy[T]], middleResult: util.List[T], params: util.Map[Any, Any]): util.List[T] = {
    var dataFrame: DataFrame = null
    val func = params.get("table").asInstanceOf[(RDD[String]) => SQLContext]
    params.put("sql",(rdd:RDD[String])=>{
      val sqlContext = func(rdd)
      dataFrame = sqlContext.sql(sql)
      dataFrame
    })
    middleResult
  }
}

上面的代碼就完成了一個(gè)SQL模塊。那如果我們要完成一個(gè)自定義的.map函數(shù)呢？可類似下面的實(shí)現(xiàn)：

abstract class MapCompositor[T,U] extends Compositor[T]{
  private var _configParams: util.List[util.Map[Any, Any]] = _
  val logger = Logger.getLogger(classOf[SQLCompositor[T]].getName)

  override def initialize(typeFilters: util.List[String], configParams: util.List[util.Map[Any, Any]]): Unit = {
    this._configParams = configParams
  }

  
  override def result(alg: util.List[Processor[T]], ref: util.List[Strategy[T]], middleResult: util.List[T], params: util.Map[Any, Any]): util.List[T] = {
    val dstream = middleResult(0).asInstanceOf[DStream[String]]
    val newDstream = dstream.map(f=>parseLog(f))
    List(newDstream.asInstanceOf[T])
  }
  def parseLog(line:String): U
}

class YourCompositor[T,U] extends MapCompositor[T,U]{

 override def parseLog(line:String):U={
     ....your logical
  }
}

同理你可以實(shí)現(xiàn)filter,repartition等其他函數(shù)。

總結(jié)

該方式提供了一套更為高層的API抽象,用戶只要關(guān)注具體實(shí)現(xiàn)而無需關(guān)注Spark的使用。同時(shí)也提供了一套配置化系統(tǒng)，方便構(gòu)建數(shù)據(jù)處理流程，并且復(fù)用原有的模塊，支持使用SQL進(jìn)行數(shù)據(jù)處理。

廣告

這個(gè)只是我們大系統(tǒng)的一小部分，愿意和我們一起進(jìn)一步完善該系統(tǒng)么？歡迎加入我們(請(qǐng)私信我)