代码提交

eagleEye-flink_kafka/src/main/java/com/flink/achieve/doris/SimiFriendsAchi.java

@@ -7,7 +7,6 @@ import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
@@ -126,8 +125,8 @@ public class SimiFriendsAchi extends SourceCommonBase implements Serializable {
 							InitiateFriendRequestReqDto jsonReqDto = JSONObject.parseObject(reqBody,
 									new TypeReference<InitiateFriendRequestReqDto>() {
 									});
-							List<String> cidList = Arrays.asList(jsonReqDto.getCid(), jsonReqDto.getFriendCid());
-							logger.info("统计条件 cid:{},friendCid:{}", jsonReqDto.getCid(), jsonReqDto.getFriendCid());
+//							List<String> cidList = Arrays.asList(jsonReqDto.getCid(), jsonReqDto.getFriendCid());
+//							logger.info("统计条件 cid:{},friendCid:{}", jsonReqDto.getCid(), jsonReqDto.getFriendCid());
 							
 							String cid = jsonReqDto.getCid();
 							String friendCid = jsonReqDto.getFriendCid();
@@ -140,7 +139,7 @@ public class SimiFriendsAchi extends SourceCommonBase implements Serializable {
 								List<SimiFriends> friendsList = JSONObject.parseObject(friendCidInfo,
 										new TypeReference<List<SimiFriends>>() {
 								});
-								logger.info("统计条件 cid:{},total:{}", cid, friendsList.size());
+//								logger.info("统计条件 cid:{},total:{}", cid, friendsList.size());
 								List<FriendsStream> friendsStreamList = new ArrayList<>();
 								for (SimiFriends friend : cidsList) {
 									FriendsStream friendsStream = new FriendsStream(
@@ -170,7 +169,9 @@ public class SimiFriendsAchi extends SourceCommonBase implements Serializable {
 								FriendsRecord friendsRecord = new FriendsRecord(
 										friendsStreamList,
 										cid,
-										friendCid
+										friendCid,
+										cidsList,
+										friendsList
 										);
 								out.collect(friendsRecord);
 							}
@@ -268,8 +269,9 @@ public class SimiFriendsAchi extends SourceCommonBase implements Serializable {
 				                        )
 				                )
 				                .count();
-				        
+				        //查询重叠cid是否有白名单
 				        List<String> overlapWhitelist = queryDorisWhitelist(overlapCids);
+				        //查询
 				        
 //				        String str = "用户分组统计条数: " + userGroupCounts+"交集数量: " + overlapCids.size()+"共享Friend交集: "+overlapCids+"共享Friend thirdId交集: " + validIdCount+"白名单cid条数:"+overlapWhitelist.size();
 				        GenericRowData row = new GenericRowData(fieldsTotal.length);
@@ -294,9 +296,11 @@ public class SimiFriendsAchi extends SourceCommonBase implements Serializable {
 	
 	public static class FriendsRecord {
 		public List<FriendsStream> friendsStreamList;
+		public List<SimiFriends> cidsList;
+		public List<SimiFriends> friendsList;
 		public String cid;
 		public String friendCid;
-        public FriendsRecord(List<FriendsStream> friendsStreamList, String cid,String friendCid) {
+        public FriendsRecord(List<FriendsStream> friendsStreamList, String cid,String friendCid,List<SimiFriends> cidsList,List<SimiFriends> friendsList) {
             this.friendsStreamList = friendsStreamList;
             this.cid = cid;
             this.friendCid = friendCid;
@@ -319,6 +323,18 @@ public class SimiFriendsAchi extends SourceCommonBase implements Serializable {
 		public void setFriendCid(String friendCid) {
 			this.friendCid = friendCid;
 		}
+		public List<SimiFriends> getCidsList() {
+			return cidsList;
+		}
+		public void setCidsList(List<SimiFriends> cidsList) {
+			this.cidsList = cidsList;
+		}
+		public List<SimiFriends> getFriendsList() {
+			return friendsList;
+		}
+		public void setFriendsList(List<SimiFriends> friendsList) {
+			this.friendsList = friendsList;
+		}
 	}
 	
 	public List<String> queryDorisWhitelist(Set<String> cidSet) {