StreamingPro 提供API实现自定义功能前言
前言
最近给StreamingPro提供了两个新的模块，一个是streamingpro-manager,一个是streamingpro-api。 streamingpro-manager主要是提供一个部署，管理Spark任务的web服务。streamingpro则是增强了定制StreamingPro的能力。当然，还有就是对原有功能的增强，比如StreamingPro SQL Server 支持异步导出数据，使得交互式查询中，如果有海量结果需要返回成为可能。
虽然我们希望能够使用SQL(加上UDF函数)完成所有的任务，但是现实往往没有那么理想。为了能够照顾灵活性，我们提供了三种方式让用户更好的使用StreamingPro完成自己的需求。
三种方案简介
通过添加UDF函数。
UDF函数是可以直接在SQL中使用的。算是一个比较自然的增强方案。具体做法是：
首先，在配置文件添加一个配置，
"udf_register": {
"desc": "测试",
"strategy": "refFunction",
"algorithm": [],
"ref": [],
"compositor": [
{
"name": "sql.udf",
"params": [
{
"analysis": "streaming.core.compositor.spark.udf.func.Functions"
}
]
}
]
}udf_register, analysis等都可以自定义命名，最好是取个有意义的名字，方便管理。
streaming.core.compositor.spark.udf.func.Functions包含了你开发的UDF函数。比如我要开发一个mkString udf函数：
object Functions {
def mkString(uDFRegistration: UDFRegistration) = {
uDFRegistration.register("mkString", (sep: String, co: mutable.WrappedArray[String]) => {
co.mkString(sep)
})
}
}之后就可以在你的Job的ref标签上引用了
{
"scalamaptojson": {
"desc": "测试",
"strategy": "spark",
"algorithm": [],
"ref": ['udf_register'],你对应的任务就可以直接使用mkString函数了。
支持了script
在配置文件中，如果能嵌入一些脚本，在特定场景里也是很方便的，这样既不需要编译啥的了。截止到这篇发布为止,支持脚本的有：
Spark 1.6.+:
* 批处理Spark 2.+:
* 批处理
* Spark Streaming处理参看我文章StreamingPro 可以暴露出原生API给大家使用,http://www.jianshu.com/p/b33c36cd3481。但是script依然会有些问题，尤其是如果你使用需要sqlContext的脚本，因为分布式的问题，会产生一些不可预期bug。同时脚本写起来也不方便，不太好利用IDE。
通过标准API
现在我们提供了API,可以定制任何你要的环节，并且和其他现有的组件可以很好的协同，当然，你也可以使用原始的Compositor接口，实现 非常高级的功能。目前支持的版本和类型有：
Spark 2.+:
* 批处理
* Spark Streaming处理这里有个spark streaming的例子，我想先对数据写代码处理，然后再接SQL组件，然后再进行存储（存储我也可能想写代码）
{
"scalamaptojson": {
"desc": "测试",
"strategy": "spark",
"algorithm": [],
"ref": [
],
"compositor": [
{
"name": "stream.sources",
"params": [
{
"format": "socket",
"outputTable": "test",
"port": "9999",
"host": "localhost",
"path": "-"
}
]
},
{
"name": "stream.script.df",
"params": [
{
"clzz": "streaming.core.compositor.spark.api.example.TestTransform",
"source": "-"
}
]
},
{
"name": "stream.sql",
"params": [
{
"sql": "select * from test2",
"outputTableName": "test3"
}
]
},
{
"name": "stream.outputs",
"params": [
{
"clzz": "streaming.core.compositor.spark.api.example.TestOutputWriter",
"inputTableName": "test3"
}
]
}
],
"configParams": {
}
}
}要实现上面的逻辑，首先是创建一个项目，然后引入下面的依赖：
<dependency>
<groupId>streaming.king</groupId>
<artifactId>streamingpro-api</artifactId>
<version>0.4.15-SNAPSHOT</version>
</dependency>这个包目前很简单，只有两个接口：
//数据处理
trait Transform {
def process(sQLContext: SQLContext, contextParams: Map[Any, Any], config: Map[String, String]): Unit
}
//数据写入
trait OutputWriter {
def write(df: DataFrame, contextParams: Map[Any, Any], config: Map[String, String]): Unit
}以数据处理为例，只要实现Transform接口，就可以通过stream.script.df 模块进行配置了。
{
"name": "stream.script.df",
"params": [
{
"clzz": "streaming.core.compositor.spark.api.example.TestTransform",
"source": "-"
}
]
},同样，我们也对输出进行了编程处理。
下面是TestTransform的实现：
class TestTransform extends Transform {
override def process(sQLContext: SQLContext, contextParams: Map[Any, Any], config: Map[String, String]): Unit = {
sQLContext.sql("select * from test").createOrReplaceTempView("test2")
}
}TestOutputWriter也是类似的：
class TestOutputWriter extends OutputWriter {
override def write(sQLContext: SQLContext, contextParams: Map[Any, Any], config: Map[String, String]): Unit = {
sQLContext.sparkSession.table(config("inputTableName")).show(100)
}
}contextParams 是整个链路传递的参数，大家可以忽略。config则是配置参数，比如如上面配置中的source参数，clzz参数等。另外这些参数都是可以通过启动脚本配置和替换的，参看如何在命令行中指定StreamingPro的写入路径