Spark Streaming + Spark SQL 实现配置化ETL流程

• 构建StreamingContext
  
• 设置checkpoint
  
• 链接数据源
  
• 各种transform
  
• foreachRDD 输出
  

{

  "test": {
    "desc": "测试",
    "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": {
    }
  }
}

1. 从Kafka消费数据
   
2. 将Kafka数据转化为表
   
3. 通过SQL进行处理
   
4. 打印输出

1. 配置化
   
2. 支持多Job配置
   
3. 支持各种数据源模块
   
4. 支持通过SQL完成数据处理
   
5. 支持多种输出模块
   

1. 动态添加或者删除job更新，而不用重启Spark Streaming
   
2. 支持Storm等其他流式引擎
   
3. 更好的多job互操作
   

1. Spark Streaming 定义为一个App
   
2. 每个Action定义为一个Job.一个App可以包含多个Job
   

{

  "job1": {
    "desc": "测试",
    "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"：{
   ........
 } 
}

• strategy 用来定义如何组织 compositor,algorithm, ref 的调用关系
  
• algorithm作为数据来源
  
• compositor 数据处理链路模块。大部分情况我们都是针对该接口进行开发
  
• ref 是对其他job的引用。通过配合合适的strategy，我们将多个job组织成一个新的job
  
• 每个组件( compositor,algorithm, strategy) 都支持参数配置
  

class SQLCompositor[T] extends Compositor[T] {

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

//策略引擎ServiceFrameStrategy 会调用该方法将配置传入进来
  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
  }

//执行的主方法，大体是从上一个模块获取SQLContext(已经注册了对应的table),
//然后根据该模块的配置，设置查询语句，最后得到一个新的dataFrame.
// middleResult里的T其实是DStream,我们会传递到下一个模块，Output模块
//params参数则是方便各个模块共享信息，这里我们将对应处理好的函数传递给下一个模块
  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
  }
}

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
  }
}