原文：http://zqhxuyuan.github.io/2017/09/04/2017-09-04-StreamingPro/
StreamingPro支持Spark、SparkStreaming、SparkStruncture、Flink。入口类都是统一的StreamingApp。
object StreamingApp {
def main(args: Array[String]): Unit = {
val params = new ParamsUtil(args)
require(params.hasParam("streaming.name"), "Application name should be set")
PlatformManager.getOrCreate.run(params)
}
}通过streaming.platform可以指定不同的运行平台。当然，不同的运行引擎的jar包也不同。
SHome=/Users/allwefantasy/streamingpro
./bin/spark-submit --class streaming.core.StreamingApp \
--master local[2] \
--name test \
$SHome/streamingpro-spark-2.0-0.4.15-SNAPSHOT.jar \
-streaming.name test \
-streaming.platform spark_streaming \
-streaming.job.file.path file://$SHome/spark-streaming.json
bin/flink run -c streaming.core.StreamingApp \
/Users/allwefantasy/streamingpro/streamingpro.flink-0.4.14-SNAPSHOT-online-1.2.0.jar \
-streaming.name god \
-streaming.platform flink_streaming \
-streaming.job.file.path file:///Users/allwefantasy/streamingpro/flink.json
jar包会被用来加载不同的Runtime。Runtime运行的映射关系定义在PlatformManager的platformNameMapping变量中。
Runtime是一个接口，最主要的是startRuntime方法和params方法。后面我们把Runtime叫做执行引擎。
trait StreamingRuntime {
def startRuntime: StreamingRuntime
def destroyRuntime(stopGraceful: Boolean, stopContext: Boolean = false): Boolean
def streamingRuntimeInfo: StreamingRuntimeInfo
def resetRuntimeOperator(runtimeOperator: RuntimeOperator)
def configureStreamingRuntimeInfo(streamingRuntimeInfo: StreamingRuntimeInfo)
def awaitTermination
def startThriftServer
def startHttpServer
def params: JMap[Any, Any]
}StreamingPro本质上还是通过spark-submit运行。框架的整体运行流程在PlatformManager的run方法中。主要的步骤有：
- 设置配置信息
- 根据反射机制，创建并获取运行时环境
- 获取dispatcher以及所有的strategies
- 启动REST服务、Thrift服务、注册ZK（可选）
- 启动执行引擎，并等待作业完成
关于Dispatcher、Strategy的概念，参考作者的ServiceframeworkDispatcher项目。
反射创建执行引擎，调用的是对应Object类的getOrCreate方法，并传入params参数，最后实例化为StreamingRuntime。
def platformNameMapping = Map[String, String](
SPAKR_S_S -> "streaming.core.strategy.platform.SparkStructuredStreamingRuntime",
SPAKR_STRUCTURED_STREAMING -> "streaming.core.strategy.platform.SparkStructuredStreamingRuntime",
FLINK_STREAMING -> "streaming.core.strategy.platform.FlinkStreamingRuntime",
SPAKR_STREAMING -> "streaming.core.strategy.platform.SparkStreamingRuntime",
SPARK -> "streaming.core.strategy.platform.SparkRuntime"
)注意：StreamingPro的Runtime只是Spark作业的执行引擎，具体根据配置文件加载策略是ServiceframeworkDispatcher的工作。
假设我们定义了下面的一个配置文件，由于采用了shortName，需要定义一个ShortNameMapping
{
"convert-multi-csv-to-json": {
"desc": "测试",
"strategy": "spark",
"algorithm": [
{
"name": "testProcessor"
}
],
"ref": [],
"compositor": [
{
"name": "testCompositor"
}
],
"configParams": {
}
}
}DefaultShortNameMapping的定义如下。这样配置文件中的spark就和ServiceframeworkDispatcher的加载过程对应起来了。
class DefaultShortNameMapping extends ShortNameMapping {
private val compositorNameMap: Map[String, String] = Map[String, String](
"spark" -> "serviceframework.dispatcher.test.DefaultStrategy",
"testProcessor" -> "serviceframework.dispatcher.test.TestProcessor",
"testCompositor" -> "serviceframework.dispatcher.test.TestCompositor"
)
override def forName(shortName: String): String = {
if (compositorNameMap.contains(shortName)) compositorNameMap(shortName)
else shortName
}
}ServiceframeworkDispatcher的核心是StrategyDispatcher，这个类在创建的时候，会读取配置文件。
然后解析配置文件中的strategy、algorithm(processor)、ref、compositor、configParams等配置项，并构造对应的对象。
ServiceframeworkDispatcher是一个模块组合框架，它主要定义了Compositor、Processor、Strategy三个接口。
Strategy接口包含了processor、ref、compositor，以及初始化和result方法。
trait Strategy[T] extends ServiceInj{
def processor:JList[Processor[T]]
def ref:JList[Strategy[T]]
def compositor:JList[Compositor[T]]
def name:String
def initialize(name:String,alg:JList[Processor[T]],ref:JList[Strategy[T]],com:JList[Compositor[T]],params:JMap[Any,Any])
def result(params:JMap[Any,Any]):JList[T]
def configParams:util.Map[Any, Any]
def stop = {}
}Strategy策略的初始化需要算法、引用、组合器，以及配置信息，对应的方法是StrategyDispatcher的createStrategy方法。
注意下面的initialize方法，createAlgorithms和createCompositors初始化时
会读取params配置，这是一个嵌套了Map的列表：JList[JMap[String, Any]]。
def createStrategy(name: String, desc: JMap[_, _]): Option[Strategy[T]] = {
if (_strategies.contains(name)) return None;
// 实例化策略，如果有shortName，则先获取fullName，并通过Class.forName实例化具体的策略类
val strategy = Class.forName(shortNameMapping.forName(desc.get("strategy").asInstanceOf[String])).newInstance().asInstanceOf[Strategy[T]]
// 读取配置信息，并实例化为Map[Any,Any]
val configParams: JMap[Any, Any] = if (desc.containsKey("configParams")) desc.get("configParams").asInstanceOf[JMap[Any, Any]] else new java.util.HashMap()
// 初始化策略，需要创建算法、引用、组合器
strategy.initialize(name, createAlgorithms(desc), createRefs(desc), createCompositors(desc), configParams)
_strategies.put(name, strategy)
Option(strategy)
}
// 创建算法。一个策略由0个或者多个算法提供结果
private def createAlgorithms(jobJMap: JMap[String, Any]): JList[Processor[T]] = {
if (!jobJMap.contains("algorithm") && !jobJMap.contains("processor")) return new AList[Processor[T]]()
val processors = if (jobJMap.contains("algorithm")) jobJMap("algorithm") else jobJMap("processor")
processors.asInstanceOf[JList[JMap[String, Any]]].map {
alg =>
val name = shortName2FullName(alg)
val processor = Class.forName(name).newInstance().asInstanceOf[Processor[T]]
val params: JList[JMap[String, Any]] = if (alg.contains("params")) alg("params").asInstanceOf[JList[JMap[String, Any]]] else new AList[JMap[String, Any]]()
processor.initialize(name, params)
processor
}
}
// 创建组合器，可以多个，按顺序调用。有点类似过滤器链。第一个过滤器会接受算法或者策略的结果。后续的组合器就只能处理上一阶段的组合器吐出的结果
private def createCompositors(jobJMap: JMap[String, Any]): JList[Compositor[T]] = {
if (!jobJMap.contains("compositor")) return new AList()
val compositors = jobJMap.get("compositor")
compositors.asInstanceOf[JList[JMap[String, Any]]].map {
f =>
val compositor = Class.forName(shortName2FullName(f)).newInstance().asInstanceOf[Compositor[T]]
val params: JList[JMap[String, Any]] = if (f.contains("params")) f.get("params").asInstanceOf[JList[JMap[String, Any]]] else new AList[JMap[String, Any]]()
compositor.initialize(f.get("typeFilter").asInstanceOf[JList[String]], params)
compositor
}
}ServiceframeworkDispatcher的核心是StrategyDispatcher，而StrategyDispatcher的核心是其dispatch方法。
def dispatch(params: JMap[Any, Any]): JList[T] = {
findStrategies(clientType) match {
case Some(strategies) =>
strategies.flatMap { f => f.result(params) }
}
}不同执行引擎的启动方法实现不同：
class SparkRuntime(_params: JMap[Any, Any]) extends StreamingRuntime with PlatformManagerListener {
override def startRuntime: StreamingRuntime = this
var sparkSession: SparkSession = createRuntime
def createRuntime = {
//...创建SparkSession，这里会根据参数判断是否支持Hive、Carbondata
}
params.put("_session_", sparkSession) //将SparkSession放入params中
registerUDF
override def params: JMap[Any, Any] = _params
}
class SparkStreamingRuntime(_params: JMap[Any, Any]) extends StreamingRuntime with PlatformManagerListener { self =>
var streamingContext: StreamingContext = createRuntime
def createRuntime = {
//创建StreamingContext，并将SparkSession放入params中
}
override def startRuntime = {
streamingContext.start()
this
}
override def awaitTermination = streamingContext.awaitTermination()
}但真正执行StreamingPro主流程在streamingpro-commons下的SparkStreamingStrategy类。
注意：如果是spark-1.6，则streamingpro-spark下也有一个SparkStreamingStrategy类。
class SparkStreamingStrategy[T] extends Strategy[T] with DebugTrait with JobStrategy {
var _ref: util.List[Strategy[T]] = _
var _compositor: util.List[Compositor[T]] = _
var _processor: util.List[Processor[T]] = _
var _configParams: util.Map[Any, Any] = _
def result(params: util.Map[Any, Any]): util.List[T] = {
ref.foreach { r => r.result(params) } // 先执行ref
if (compositor != null && compositor.size() > 0) {
// 第一个Compositor, 产生第一个中间结果
var middleR = compositor.get(0).result(processor, ref, null, params)
// 将新的中间结果运用到下一个Compositor
// 第一个Compositor的结果运用到第二个的输入, 第二个Compositor的结果运用到第三个Compositor的输入...
// 所以不同Compositor是链式执行的
for (i <- 1 until compositor.size()) {
middleR = compositor.get(i).result(processor, ref, middleR, params)
}
middleR
} else new util.ArrayList[T]()
}
}注意：配置文件中每个Job都有一个strategy级别的configParams，ref也会使用这个全局的configParams。
它是一个Map[String, Any]的结构。每个Compositor和Processor内部也有一个params配置，这是一个数组。
实际上，全局的configParams参数会被用在Strategy、Ref/Processor和Compositor的result()方法的最后一个参数。
"compositor": [
{
"name": "testCompositor",
"params": [
{
"sql": "select avg(value) avgAge from test",
"outputTableName": "test3"
},
{
"sql": "select sum(value) sumAge from test",
"outputTableName": "test4"
}
]
}
],
接下来以读取多个数据源的Compositor实现类为例：
_configParams是在创建Compositor时初始化调用的，这是一个List[Map[String, Any]]的结构，对应了params列表配置- 如果需要替换，则会先处理配置信息
- 接着，从params中获取SparkSession（还记得之前创建Runtime时放入Map中吗？），
- 然后，执行sparkSession.read.format(xx).options(Map).load(path)
- 最后，通过df.createOrReplaceTempView创建Spark SQL的临时表，名称为
outputTable
class MultiSQLSourceCompositor[T] extends Compositor[T] with CompositorHelper {
private var _configParams: util.List[util.Map[Any, Any]] = _
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] = {
_configParams.foreach { sourceConfig =>
val name = sourceConfig.getOrElse("name", "").toString
val _cfg = sourceConfig.map(f => (f._1.toString, f._2.toString)).map { f =>
(f._1, params.getOrElse(s"streaming.sql.source.${name}.${f._1}", f._2).toString)
}.toMap
val sourcePath = _cfg("path")
val df = sparkSession(params).read.format(sourceConfig("format").toString).options(
(_cfg - "format" - "path" - "outputTable").map(f => (f._1.toString, f._2.toString))).load(sourcePath)
df.createOrReplaceTempView(_cfg.getOrElse("outputTable", _cfg.getOrElse("outputTableName", "")))
}
List()
}
}
为了支持配置的动态替换，_cfg参数会做一些处理，比如上面的s"streaming.sql.source.${name}.${f._1}"如果需要被替换，则会被替换为f._2。
下表列举了StreamingPro支持的几种替换方式。
| 配置参数 |
配置示例 |
动态传参数 |
streaming.sql.source.[name].[参数] |
"path": "file:///tmp/sample_article.txt" |
-streaming.sql.source.firstSource.path file:///tmp/wow.txt |
streaming.sql.out.[name].[参数] |
"path": "file:///tmp/sample_article.txt" |
-streaming.sql.source.firstSink.path file:///tmp/wow_20170101.txt |
streaming.sql.params.[param-name] |
"sql": "select * from test where hp_time=:today" |
-streaming.sql.params.today "20170101" |
假设有两个数据输入源和一个输出目标的配置如下：
{
"name": "batch.sources",
"params": [
{
"name":"firstSource",
"path": "file:///tmp/sample_article.txt",
"format": "com.databricks.spark.csv",
"outputTable": "article",
"header":true
},
{
"name":"secondSource",
"path": "file:///tmp/sample_article2.txt",
"format": "com.databricks.spark.csv",
"outputTable": "article2",
"header":true
}
]
},
{
"name": "batch.outputs",
"params": [
{
"name":"firstSink",
"path": "file:///tmp/sample_article.txt",
"format": "com.databricks.spark.csv",
"outputTable": "article",
"header":true
}
]
}
Source的功能是：读取输入源形成DataFrame，然后创建临时表。其他组件比如SQL也是类似的。至此StreamingPro的大致流程就分析完了。
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent