### --- 实现步骤~~~     获取数据源(input)--- flume
~~~     转化
~~~     数据源的格式：area：uid：product_id：time:  样例类中AdClick；过滤操作filter product_id ！= null
~~~     .水印Watemark、.keyBy(productId) 、.timeWindow
~~~     .aggregate(MyAggFunc,MyWindowFunc)
~~~     MyAggFunc:编写计算逻辑的代码---- 累加广告的点击次数
~~~     MyWindowFunc：apply，将结果数据向下游传递

### --- MyAggFunc：编写计算逻辑的代码--累加广告的点击次数{"yanqi_event": [{"name": "goods_detail_loading","json": {"entry": "2","goodsid": "0","loading_time": "92","action": "3","staytime": "10","showtype": "0"},"time": 1595265099584},{"name": "notification","json": {"action": "1","type": "3"},"time": 1595341087663},{"name": "ad","json": {"duration": "10","ad_action": "0","shop_id": "23","event_type": "ad","ad_type": "1","show_style": "0","product_id": "36","place": "placecampaign2_left","sort": "1"},"time": 1595276738208}],"attr": {"area": "东莞","uid": "2F10092A0","app_v": "1.1.0","event_type": "common","device_id": "1FB872-9A1000","os_type": "1.1","channel": "广宣","language": "chinese","brand": "iphone-0"}
}

### --- ODS:kafka中eventlog中；DIM:无；DWD：event.log文件说明：不规范json格式，需要转换{"data": [{"id": "6","payMethod": "meituan","payName": "美团支付","description": "美团支付","payOrder": "0","online": "-1"}],"database": "dwshow","es": 1604461572000,"id": 6,"isDdl": false,"mysqlType": {"id": "int(11)","payMethod": "varchar(20)","payName": "varchar(255)","description": "varchar(255)","payOrder": "int(11)","online": "tinyint(4)"},"old": null,"pkNames": null,"sql": "","sqlType": {"id": 4,"payMethod": 12,"payName": 12,"description": 12,"payOrder": 4,"online": -6},"table": "yanqi_payments","ts": 1604461572297,"type": "INSERT"
}

{"yanqi_event": [{"name": "goods_detail_loading","json": {"entry": "2","goodsid": "0","loading_time": "92","action": "3","staytime": "10","showtype": "0"},"time": 1595265099584},{"name": "notification","json": {"action": "1","type": "3"},"time": 1595341087663},{"name": "ad","json": {"duration": "10","ad_action": "0","shop_id": "23","event_type": "ad","ad_type": "1","show_style": "0","product_id": "36","place": "placecampaign2_left","sort": "1"},"time": 1595276738208}],"attr": {"area": "东莞","uid": "2F10092A0","app_v": "1.1.0","event_type": "common","device_id": "1FB872-9A1000","os_type": "1.1","channel": "广宣","language": "chinese","brand": "iphone-0"}
}

### --- 转换代码//对Kafka中的JSON日志进行转换val mapEventStream: DataStream[AdClick] = eventLogStream.map(x => {val jsonObj: JSONObject = JSON.parseObject(x)val attr: String = jsonObj.get("attr").toStringval attrJson: JSONObject = JSON.parseObject(attr)val area: String = attrJson.get("area").toStringval uid: String = attrJson.get("uid").toString//[{"name":"praise","json":{"id":0,"type":4,"add_time":"1597851188753","userid":0,"target":8},// "time":1595329059805}]// 此处的时间戳是毫秒val eventData: String = jsonObj.get("yanqi_event").toStringval datas: JSONArray = JSON.parseArray(eventData)val list = new java.util.ArrayList[String]()datas.forEach(x => list.add(x.toString))var productId: String = nullvar timestamp: Long = 0Llist.forEach(x => {//{"name":"ad","json":{"duration":"10","ad_action":"0","shop_id":"23","event_type":"ad","ad_type":"1",//"show_style":"0","product_id":"36","place":"placecampaign2_left","sort":"1"},"time":1595276738208}val xJson: JSONObject = JSON.parseObject(x)if (xJson.get("name").toString.equals("ad")) {val jsonData: String = xJson.get("json").toStringval jsonDatas = JSON.parseObject(jsonData)productId = jsonDatas.get("product_id").toStringtimestamp = TimeUnit.MILLISECONDS.toSeconds(xJson.get("time").toString.toLong)}})AdClick(area, uid, productId, timestamp)
})