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的写入路径