log-parse.json

{
	"XXX": {
    "desc": "XXX",
    "strategy": "streaming.core.strategy.SparkStreamingStrategy",
    "algorithm": [],
    "ref": [],
    "compositor": [
      {
        "name": "streaming.core.compositor.spark.streaming.source.KafkaStreamingCompositor",
        "params":[
          {
          "metadata.broker.list":"",
          "topics":"",
	        "auto.offset.reset":"smallest"
          }
        ]
      },
      {
        //字符串解析的配置，输出的是个Java Map
      	"name":"com.letvcloud.bigdata.spark.transform.DouyuCompositor",
      	"params": [
          {
            "conf":"http://..../conf/?conf=.../douyu-kd.json",
            "chainName":"....",
            "unSerialize":false
          }
        ]
      },
	  {
        //把 JavaMap 转化为JSON字符串
        "name": "streaming.core.compositor.spark.streaming.transformation.JavaMapToJSONCompositor", 
        "params": [
          {
          }
        ]
      },
      {
        //注册成表
        "name": "streaming.core.compositor.spark.streaming.transformation.JSONTableCompositor",
        "params": [
          {
            "tableName": "douyu_kd"
          }
        ]
      },
      {
        // 查询和计算
        "name": "streaming.core.compositor.spark.streaming.transformation.SQLCompositor",
        "params": [
          {
            "sql": "select floor(data_str/1000000) as day,floor(data_str/10000) as hour,floor(floor(data_str/100)/5)*5 as min5,stream_id,province,isp,rid as room_id,client_ip,sum(ec) as ka_num from douyu_kd group by floor(data_str/1000000),floor(data_str/10000),floor(floor(data_str/100)/5)*5,stream_id,province,isp,rid,client_ip"
          }
        ]
      },
      {
        //输出到ElasticSearch
        "name":"streaming.core.compositor.spark.streaming.output.SQLESOutputCompositor",
        "params":[
          {
            "es.nodes":"",
            "resource":"xxxx/docs",
            "es.resource":"xxxx/docs",
            "timeFormat":"yyyyMMdd"
          }
        ]
      }
    ]
  }
}