spark在风控用户团伙中的应用

投资就是价值观的变现

文章目录

引言
SparkGraphX
- 执行过程
- 后期优化
Neo4j
- 存储
- 结构
- 使用
SparkStreaming
- 秒级实时关系存储
- 纬度关联+存储
Drools

引言

2020年年初业务稳步增长，风控方面遇到了挑战，作为C2C(类似于得物、咸鱼)交易平台，难免会遇到用户团伙进行薅平台羊毛、自买自卖的行为，风控团队将用户的历史数据整合后，希望从中圈选出各个大大小小的团伙，来防止团伙内部和团伙之间进行交易，数据量大，纬度复杂，经过几个周的磨合尝试，以及和某些小额贷款公司的交流，终于确定了一套完整的团伙发现体系，最终走向数据的实时化，其中spark的GraphX和streaming占据重要位置。接下来简述1.0版本的应用。

SparkGraphX

选择使用SparkGraphX有两点：1、数据量大，使用普通python或者java程序实现不了，顺便用spark直接读取hive的数据；
2、使用spark强大的图计算功能，进行连通图的计算等等

执行过程

1、获取数据，从hive中获取数据转化为DF(点-关系)

    val sql =s"""|select uid, fid, type from dw.user_info_xx_d where partition_date=${date}""".stripMarginval matedata: RDD[Row] = hc.sql(sql).rddmatedata.map(elem => {val lineType = elem.getAs[String](2)println(lineType)lineType match{case "充值" => (elem.getAs[Long](1).toString, elem.getAs[String](0), elem.getAs[String](2), "BindAccount")case "支付" => (elem.getAs[Long](1).toString, elem.getAs[String](0), elem.getAs[String](2), "BindAccount")case "认证" => (elem.getAs[Long](1).toString, elem.getAs[String](0), elem.getAs[String](2), "RealnameVerify")...case _ => {logger.error("========问题数据："+elem+"================")null}}})

2、数据格式化为标准的点和边， (点集合，关系集合)

//数据格式化为点边关系
val node_rels = format_node_rels(uid_tid_rels_type)//数据格式化为GrphX标准点和边
val vertexRDD: RDD[(VertexId, PartitionID)] = node_rels._1
val edgeRDD: RDD[Edge[PartitionID]] = node_rels._2.map(elem => Edge(elem._1, elem._2._1, elem._2._2))def format_node_rels(rdd: RDD[(String, String, String, String)]): (RDD[(VertexId, PartitionID)], RDD[(VertexId, (VertexId, PartitionID))]) = {val uid_nodes: RDD[(VertexId, PartitionID)] = rdd.map(elem => (elem._1.toLong, 1)).reduceByKey(_ + _).map(elem => (elem._1, 1))//通过uid的相同关系，将uid们连接起来val tid_nodes: RDD[(VertexId, (VertexId, PartitionID))] = rdd.map(elem => (elem._2, elem._1.toLong)).groupByKey().flatMap(elem => {val list = elem._2val min = list.minprintln(list)list.map(index => {((min, index), 1)})}).filter(elem => elem._1._1 != elem._1._2).reduceByKey(_ + _).map(elem => (elem._1._1, (elem._1._2, 1)))(uid_nodes, tid_nodes)}

3、构建图

 val graph: Graph[PartitionID, PartitionID] = Graph(uid_nodes, tid_nodes)

4、构建联通关系

val gang: RDD[(VertexId, Iterable[VertexId])] = connected(graph, vertexRDD)/*** 联通图* @param graph* @param vertexRDD*/
def connected(graph: Graph[PartitionID, PartitionID], vertexRDD: RDD[(VertexId, PartitionID)]): RDD[(VertexId, Iterable[VertexId])] = {//获得连通图val ccGraph: Graph[VertexId, PartitionID] = graph.connectedComponents()val res: RDD[(VertexId, Iterable[VertexId])] = ccGraph.vertices.join(vertexRDD).map {case (uid, (gangid, num)) => (gangid, uid)}.groupByKey()res}

5、结果存储

gang.foreachPartition(rdd => {rdd.foreach(elem => {val gandId = elem._1val list = elem._2val size = list.sizeif (size > 1) {list.foreach(uid => {//用户id，所在联通图id，联通图大小println(uid, gandId, size)})})})
})

后期优化

属于2期的优化，先对不同的关系进行加权，比如两个人同一wifi、同一经纬度，这个时候权重特别低；两个人用过同一设备、同一支付账号，这钟权重比较高。再通过对联通图使用Louvain进行训练，将1期中难以拆分的大联通图进行关系优化。

Neo4j

neo4j是一个开源的图数据库（集群版不开源，按节点收费，贼贵，单节点免费），同一业务拆分了一下，能满足当前的数据需求。只做查询，不做计算，之前也考虑过计算，效率低，内存消耗高，最终选择了SparkGraphX作为计算。

存储

存储分两部分：存量数据的初始化和增量数据，具体怎么实现，存量数据的初始化看我另一篇关于Neo4j的博客，增量数据在下面的代码中有。

结构

存储结构uid–>共同关系<–uid：
1002–实名认证–>小明<–实名认证–1044或者1002–支付账号–>zfbxxxxdfs1000<–提现账号–1044

使用

提供可视化的用户关系查询，经常会遇上不知道两个人为什么被关联到一个团伙里
查询两个人之间的10度关系

MATCH p=(from:User{uid:'1002'})-[*..10]-(to:User{uid:'1044'}) RETURN p limit 100

SparkStreaming

sparkStreaming(有条件的用flink)做两件事：

秒级实时关系存储

数据源：binlog(kafka)
数据存储：neo4j、kafka、redis
处理过程：(1)、普通数据，A用户的a特征，直接存储到neo4j、kafka、redis
(2)、多重join的数据，需要两个topic进行join的，没有用直接用join，用的是窗口函数+groupByKey
窗口大小是4s钟，滑动间隔是2s钟(根据业务本身延时和数据本身延时决定，因为此动作触发后，app会有几秒的等待时间，以及多个动作后的结果，保证风控数据到位后，业务才会去调用，否则，当羊毛党都薅完羊毛了，才检测到有风险，岂不是有些扯淡，所以有些动作比较卡，不是数据拥挤导致的，也许就说程序员在里面写的sleep(n)，哈哈哈)。比如uid和支付id的关联关系，需要两个表，表中关键字段分表是，（order_id, uid）(order_id, pay_id)
，先将两个topic合流，然后调出相同字段

 .map(elem => (((elem._1, elem._2), elem._3), 1)) // ((提现表关联 ，order_id)，uid/pay_id).reduceByKeyAndWindow((x:Int, y:Int)=>x+y, Seconds(30), Seconds(15)).map(elem=>(elem._1._1, elem._1._2)) //.groupByKey() // (统一表名 ，订单id) => (uid+pay_id).map(elem=>{//整理数据，除了提现表关联，还有很多关联关系}).foreach(存储)

neo4j

  //更新关系private def relationShip(session: Session, uid: Int, flag_label:String, flag_type:String, fuid: String): Any = {val sql =  s"""|match (p1:User{userId:$uid})-[r:$flag_type]->(p2:$flag_label{flagId:"$fuid"}) return p1.userId as uid""".stripMarginval result: StatementResult = session.run(sql)if (!result.hasNext) {val sql1 =s"""|MERGE (p1:User{userId:$uid})|MERGE (p2:$flag_label{flagId:"$fuid"})""".stripMarginsession.run(sql1)val sql =s"""|MATCH (p1:User{userId:$uid}),(p2:$flag_label{flagId:"$fuid"})|MERGE (p1)-[r:$flag_type{score:1}]->(p2)|RETURN p1.userId as uid""".stripMarginval rel: StatementResult = session.run(sql)if (rel.hasNext) {val record = rel.next()val uid = record.get("uid").toStringprintln(uid)}else{System.err.println("error："+uid+flag_label+flag_type+fuid)}}}/*** 获取Driver* @return*/def getDriver(): Driver = {val url = "bolt://neo4j01:8687"val user = "user"val password = "password"val driver = GraphDatabase.driver(url, AuthTokens.basic(user, password), Config.build().withMaxIdleSessions(1000).withConnectionLivenessCheckTimeout(10, TimeUnit.SECONDS).toConfig)driver}/*** 获取Session* @param driver* @return*/def getSession(driver: Driver): Session = {val session = driver.session()session}

kafka

 def resToKafka(ssc: StreamingContext, kafkaDStreamValue: DStream[(String, String, String, String)]): Unit = {//广播KafkaSinkval kafkaProducer: Broadcast[KafkaSink[String, String]] = {val kafkaProducerConfig = {val p = new Properties()p.put("group.id", "realtime")p.put("acks", "all")p.put("retries ", "1")p.setProperty("bootstrap.servers", GetPropKey.brokers)p.setProperty("key.serializer", classOf[StringSerializer].getName)p.setProperty("value.serializer", classOf[StringSerializer].getName)p}ssc.sparkContext.broadcast(KafkaSink[String, String](kafkaProducerConfig))}//写入KafkakafkaDStreamValue.foreachRDD(rdd => {if (!rdd.isEmpty()) {rdd.foreachPartition(partition => {partition.foreach(elem => {val flag_label = elem._3if (!flag_label.equals("null")) {val auth_info = elem._1val uid = elem._2.toIntval flag_type = elem._4val value = dataJson(uid, auth_info, flag_label, flag_type)kafkaProducer.value.send("risk_user_auth_info", value)}})})}})}/*** json格式化** @param uid* @param fid* @param flag_label* @param flag_type* @return*/def dataJson(uid: Int, fid: String, flag_label: String, flag_type: String): String = {val map = Map("user_id" -> uid,"flag_id" -> fid,"flag_label" -> flag_label,"flag_type" -> flag_type)JSONObject.apply(map).toString()}

纬度关联+存储

数据源：risk_user_auth_info(1中写入kafka的延时队列）
数据存储：redis
处理过程：从延时队列中取出数据，到redis中查询出关系，归结到父节点(迭代版本中加入了算法进行合并和拆分图)，再存入redis

Drools

更新中。。。