SpringBoot整合Flink，实现一个用户个性化推荐系统

通过SpringBoot联合Flink的高性能流处理本事，开发者可以构建出高效、可靠的及时数据处理办理方案。这些应用不仅可以或许提高企业的运营服从，还能增强竞争力。

  应用场景

  1. 推荐系统：

  

• 根据用户的及时行为生成个性化推荐。
  
  
• 更新用户画像，以适应不断变化的兴趣和偏好。
  
  

  2. 变乱驱动架构：

  

• 在微服务架构中处理跨服务的消息通报，确保系统的低耽误和高吞吐量。
  
  
• 处理来自不同泉源的变乱，并根据业务规则执行相应的操作。
  
  

  3. 欺诈检测：

  

• 及时监控金融交易或网络活动，辨认异常模式并触发警报。
  
  
• 提供快速相应机制，减少欺诈损失。
  
  

  4. 流式数据分析：

  

• 及时分析传感器数据，如物联网设备的数据流。
  
  
• 监控和分析用户行为数据，提供即时反馈。
  
  

  5. 日志处理：

  

• 及时收集、解析和聚合日志数据，帮助进行性能调优和故障清除。
  
  
• 提供及时陈诉和仪表板，便于监控系统状态。
  
  

  6. 及时报表：

  

• 生成及时销售陈诉、市场趋势分析等。
  
  
• 支持决议订定者获得最新的业务洞察力。
  
  

  7. 供应链管理：

  

• 及时跟踪库存水平、订单状态和物流信息。
  
  
• 主动调整生产计划以满足需求变化。
  
  

  8. 交际媒体分析：

  

• 及时监测舆情，分析公众对特定话题的见解。
  
  
• 提供及时的情感分析，帮助企业相识品牌形象。
  
  

  9. 网络安全：

  

• 及时监控网络流量，检测埋伏的安全威胁。
  
  
• 快速相应攻击，保护企业免受损害。
  
  

  代码演示

  

我们现在用简朴代码例子演示怎样根据用户的及时行为生成个性化推荐。以下代码将展示怎样从Kafka读取用户行为数据，并使用Flink进行及时处理，最后生成个性化的推荐效果并将其写回到另一个Kafka主题中。

  1. 依赖项

  

在pom.xml中添加须要的依赖项：

1. <dependencies>
2.     <!-- Spring Boot Starter Web -->
3.     <dependency>
4.         <groupId>org.springframework.boot</groupId>
5.         <artifactId>spring-boot-starter-web</artifactId>
6.     </dependency>
8.     <!-- Flink dependencies -->
9.     <dependency>
10.         <groupId>org.apache.flink</groupId>
11.         <artifactId>flink-java</artifactId>
12.         <version>1.14.6</version>
13.     </dependency>
14.     <dependency>
15.         <groupId>org.apache.flink</groupId>
16.         <artifactId>flink-streaming-java_2.12</artifactId>
17.         <version>1.14.6</version>
18.     </dependency>
19.     <dependency>
20.         <groupId>org.apache.flink</groupId>
21.         <artifactId>flink-clients_2.12</artifactId>
22.         <version>1.14.6</version>
23.     </dependency>
24.     <dependency>
25.         <groupId>org.apache.flink</groupId>
26.         <artifactId>flink-connector-kafka_2.12</artifactId>
27.         <version>1.14.6</version>
28.     </dependency>
30.     <!-- JSON processing with Jackson -->
31.     <dependency>
32.         <groupId>com.fasterxml.jackson.core</groupId>
33.         <artifactId>jackson-databind</artifactId>
34.     </dependency>
35. </dependencies>

复制代码

2. 界说数据模型

  

界说用户行为数据和推荐效果的数据模型。

1. package com.example.demo.model;
3. publicclass UserBehavior {
4.     private String userId;
5.     private String productId;
6.     private String action; // e.g., "view", "click", "purchase"
7.     privatelong timestamp;
9.     // Getters and setters
10.     public String getUserId() {
11.         return userId;
12.     }
14.     public void setUserId(String userId) {
15.         this.userId = userId;
16.     }
18.     public String getProductId() {
19.         return productId;
20.     }
22.     public void setProductId(String productId) {
23.         this.productId = productId;
24.     }
26.     public String getAction() {
27.         return action;
28.     }
30.     public void setAction(String action) {
31.         this.action = action;
32.     }
34.     public long getTimestamp() {
35.         return timestamp;
36.     }
38.     public void setTimestamp(long timestamp) {
39.         this.timestamp = timestamp;
40.     }
41. }

复制代码

1. package com.example.demo.model;
3. import java.util.List;
5. publicclass RecommendationResult {
6.     private String userId;
7.     private List<String> recommendedProducts;
9.     // Getters and setters
10.     public String getUserId() {
11.         return userId;
12.     }
14.     public void setUserId(String userId) {
15.         this.userId = userId;
16.     }
18.     public List<String> getRecommendedProducts() {
19.         return recommendedProducts;
20.     }
22.     public void setRecommendedProducts(List<String> recommendedProducts) {
23.         this.recommendedProducts = recommendedProducts;
24.     }
25. }

复制代码

3. 编写Flink Job

  

创建一个简朴的Flink作业，它从Kafka读取用户行为数据，计算每个用户的热门产品，并将推荐效果写入另一个Kafka主题。

1. package com.example.demo;
3. import com.example.demo.model.RecommendationResult;
4. import com.example.demo.model.UserBehavior;
5. import org.apache.flink.api.common.functions.MapFunction;
6. import org.apache.flink.api.java.tuple.Tuple2;
7. import org.apache.flink.streaming.api.datastream.DataStream;
8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
9. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
10. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
11. import org.apache.flink.api.common.serialization.SimpleStringSchema;
12. import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
14. import java.util.ArrayList;
15. import java.util.HashMap;
16. import java.util.List;
17. import java.util.Map;
18. import java.util.Properties;
20. publicclass RecommendationJob {
22.     public static void main(String[] args) throws Exception {
23.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
25.         Properties properties = new Properties();
26.         properties.setProperty("bootstrap.servers", "localhost:9092");
27.         properties.setProperty("group.id", "test");
29.         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("user-behavior-topic", new SimpleStringSchema(), properties);
30.         kafkaConsumer.setStartFromEarliest();
32.         ObjectMapper objectMapper = new ObjectMapper();
33.         DataStream<UserBehavior> userBehaviors = env.addSource(kafkaConsumer)
34.                 .map((MapFunction<String, UserBehavior>) value -> objectMapper.readValue(value, UserBehavior.class));
36.         DataStream<Tuple2<String, Map<String, Integer>>> productCountsPerUser = userBehaviors
37.                 .filter(behavior -> behavior.getAction().equals("purchase"))
38.                 .keyBy(UserBehavior::getUserId)
39.                 .flatMap((FlatMapFunction<UserBehavior, Tuple2<String, Map<String, Integer>>>) (value, out) -> {
40.                     Map<String, Integer> productCounts = new HashMap<>();
41.                     if (productCounts.containsKey(value.getProductId())) {
42.                         productCounts.put(value.getProductId(), productCounts.get(value.getProductId()) + 1);
43.                     } else {
44.                         productCounts.put(value.getProductId(), 1);
45.                     }
46.                     out.collect(Tuple2.of(value.getUserId(), productCounts));
47.                 })
48.                 .keyBy(Tuple2::f0)
49.                 .reduce((t1, t2) -> {
50.                     t1.f1.forEach((productId, count) -> t2.f1.merge(productId, count, Integer::sum));
51.                     return t1;
52.                 });
54.         DataStream<RecommendationResult> recommendations = productCountsPerUser
55.                 .map((MapFunction<Tuple2<String, Map<String, Integer>>, RecommendationResult>) value -> {
56.                     RecommendationResult result = new RecommendationResult();
57.                     result.setUserId(value.f0);
58.                     List<String> recommendedProducts = new ArrayList<>(value.f1.keySet());
59.                     result.setRecommendedProducts(recommendedProducts.subList(0, Math.min(5, recommendedProducts.size())));
60.                     return result;
61.                 });
63.         FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>("recommendations-topic",
64.                 (RecommendationResult recommendation) -> objectMapper.writeValueAsString(recommendation),
65.                 properties);
67.         recommendations.map(ObjectMapper::writeValueAsString).addSink(kafkaProducer);
69.         env.execute("Recommendation Job");
70.     }
71. }

复制代码

4. 创建Spring Boot Application

  

创建一个简朴的Spring Boot应用步伐来启动Flink作业。

1. package com.example.demo;
3. import org.springframework.boot.SpringApplication;
4. import org.springframework.boot.autoconfigure.SpringBootApplication;
5. import org.springframework.context.annotation.Bean;
7. @SpringBootApplication
8. publicclass DemoApplication {
10.     public static void main(String[] args) {
11.         SpringApplication.run(DemoApplication.class, args);
12.     }
14.     @Bean
15.     public Runnable flinkRunner() {
16.         return () -> {
17.             try {
18.                 RecommendationJob.main(new String[]{});
19.             } catch (Exception e) {
20.                 thrownew RuntimeException(e);
21.             }
22.         };
23.     }
24. }

复制代码

测试效果

     “   

请确保你的环境中已经设置好了Kafka和其他须要的服务，以便代码可以或许正常运行。

   

我本地的Kafka集群正在运行，并且有两个主题：user-behavior-topic 和 recommendations-topic。

  

• 向user-behavior-topic发送消息：
  1. {"userId": "user1", "productId": "productA", "action": "purchase", "timestamp": 1672531200000}
  2. {"userId": "user1", "productId": "productB", "action": "purchase", "timestamp": 1672531200000}
  3. {"userId": "user1", "productId": "productC", "action": "purchase", "timestamp": 1672531200000}
  4. {"userId": "user2", "productId": "productB", "action": "purchase", "timestamp": 1672531200000}
  5. {"userId": "user2", "productId": "productD", "action": "purchase", "timestamp": 1672531200000}

  复制代码
• 检查recommendations-topic中的消息：
  1. {"userId":"user1","recommendedProducts":["productA","productB","productC"]}
  2. {"userId":"user2","recommendedProducts":["productB","productD"]}

  复制代码

  关注我，送Java福利

1. /**
2.  * 这段代码只有Java开发者才能看得懂！
3.  * 关注我微信公众号之后，
4.  * 发送:"666"，
5.  * 即可获得一本由Java大神一手面试经验诚意出品
6.  * 《Java开发者面试百宝书》Pdf电子书
7.  * 福利截止日期为2025年01月22日止
8.  * 手快有手慢没！！！
9. */
10. System.out.println("请关注我的微信公众号：");
11. System.out.println("Java知识日历");

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。