电商平台的用户行为比较频繁和复杂，由于用户量的巨大，可以收集到大量的用户行为数据。因此可以通过大数据技术对其进行深入挖掘和分析，从而得到感兴趣的指标来增强对风险的控制。
此外，用户行为的数据多样，包括用户行为习惯数据和业务行为数据两大类。其中用户行为数据包括用户的登录、上线及时长点击浏览页面，可以进行流量统计和热门商品的统计从而深挖用户的特征；而业务行为数据是用户对每个业务所作的操作，分为两类，一类是能够明显地表现出用户兴趣的行为，我们可以对数据进行深入分析，得到用户画像；另一类则是常规的业务操作，但需要关注一些异常状况来做好风控。项目的主要模块分为热门统计、偏好统计、风险控制。
热门统计：
我们首先准备了一个数据集，包含淘宝用户某一天随机一百万用户的所有行为，包括用户ID、商品ID、商品类目ID、行为类型和时间戳等。之后我们建立了一个Maven工程，配置了flink.version的properties，添加了dependencies的项目依赖。
我们将数据文件复制到资源文件目录src/main/resources下，并从这里读取数据。为了实现“实时热门商品”的需求，我们需要每隔5分钟输出最近一小时内点击量最多的前N个商品，因此我们需要做以下几件事：
1.抽取出业务时间戳，告诉Flink框架基于业务时间做窗口
2.过滤出点击行为数据
3.按一小时的窗口大小，每5分钟统计一次，做滑动窗口聚合
4.按每个窗口聚合，输出每个窗口中点击量前N名的商品
程序主体如下所示：
case class UserBehavior(userId: Long, itemId: Long, categoryId: Int, behavior: String,
timestamp: Long)
case class ItemViewCount(itemId: Long, windowEnd: Long, count: Long)
object HotItems {
def main(args: Array[String]): Unit = {
// 创建一个 StreamExecutionEnvironment
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 设定 Time 类型为 EventTime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
// 为了打印到控制台的结果不乱序，我们配置全局的并发为 1，这里改变并发对结果正确性没有影响
env.setParallelism(1)
val stream = env
// 以 window 下为例，需替换成自己的路径
.readTextFile("YOUR_PATH\resources\UserBehavior.csv")
.map(line => {
val linearray = line.split(",")
UserBehavior(linearray(0).toLong, linearray(1).toLong, linearray(2).toInt,
linearray(3), linearray(4).toLong)
})
// 指定时间戳和 watermark
.assignAscendingTimestamps(_.timestamp * 1000)
env.execute("Hot Items Job")
}
