devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

kafka-stream流式处理示例_kafka stream流式处理

作者：知新_RL | 2024-06-23 14:20:31

赞

踩

kafka stream流式处理

   一 首生是kafka -stream 版本号问题，然后是springboot1.5.6兼容问题，发现springboot2.0不支持kafka -stream1.0.2包

  1. 第一个是窗口聚合的初始值，第二个是进行聚合的聚合器，第三个就是窗口时间，第四个是类似于序列化的东西
  2.  
  3. KTable aggregate = groupedStream.aggregate(initializer, aggregator, timeWindows, resultSerde);

下面直接依赖包  坑了很久，各种找版本

  1.  <parent>
  2.         <groupId>org.springframework.boot</groupId>
  3.         <artifactId>spring-boot-starter-parent</artifactId>
  4.         <!-- 1.5.6版本 能配置 kafka-stream 1.0.2  最新版本的流依赖版，简化很多步骤  不支持kafak的 springboot 2.0-->
  5.          <version>1.5.6.RELEASE</version>  
  6.         <relativePath/>
  7.     </parent>
  8. <dependency>
  9.             <groupId>org.apache.kafka</groupId>
  10.             <artifactId>kafka_2.12</artifactId>
  11. 		       <version>1.1.0</version>
  12.             <exclusions>
  13.                 <exclusion>
  14.                     <groupId>com.101tec</groupId>
  15.                     <artifactId>zkclient</artifactId>
  16.                 </exclusion>
  17. 		        <exclusion>
  18. 		    	<groupId>org.slf4j</groupId>
  19. 			       <artifactId>slf4j-log4j12</artifactId>
  20. 		         </exclusion>
  21. 	        </exclusions>
  22.         </dependency> 
  23.          <dependency>
  24.             <groupId>org.apache.kafka</groupId>
  25.             <artifactId>kafka-streams</artifactId>
  26.             <version>1.0.2</version>
  27.         </dependency>
  28.          <dependency>
  29.             <groupId>com.101tec</groupId>
  30.             <artifactId>zkclient</artifactId>
  31.             <version>0.10</version>
  32.         </dependency> 
  33.         <dependency>
  34.             <groupId>commons-io</groupId>
  35.             <artifactId>commons-io</artifactId>
  36.             <version>2.5</version>
  37.         </dependency>

二 最后计算是成功了， 但是因为序列化原因一直显示不出结果，所有结果都是0

下面是能用序列化方法

GenericDeserializer.java

  1. package cloud.stream.serdes;
  2.  
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import org.apache.kafka.common.errors.SerializationException;
  5. import org.apache.kafka.common.serialization.Deserializer;
  6.  
  7. import java.util.Map;
  8.  
  9. //反序列化实现
  10.  
  11. public class GenericDeserializer<T> implements Deserializer<T> {
  12.  
  13.     private ObjectMapper objectMapper = new ObjectMapper();
  14.     private Class<T>     type;
  15.     /**
  16.      * Default constructor needed by Kafka
  17.      */
  19.     public GenericDeserializer() {
  20. 	}
  21.  
  22. 	@SuppressWarnings("unchecked")
  23.     @Override
  24.     public void configure(Map<String, ?> props, boolean isKey) {
  25.     	type = (Class<T>) props.get("JsonPOJOClass");
  26.     }
  27.  
  28.     @Override
  29.     public T deserialize(String topic, byte[] bytes) {
  30.         if (bytes == null)
  31.             return null;
  32.  
  33.         T data;
  34.         try {
  35.             data = objectMapper.readValue(bytes, type);
  36.         } catch (Exception e) {
  37.             throw new SerializationException(e);
  38.         }
  39.  
  40.         return data;
  41.     }
  42.  
  43.     @Override
  44.     public void close() {
  45.  
  46.     }
  47.  
  48.  
  49. }

GenericSerializer.java

 

  1. package cloud.stream.serdes;
  2.  
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import org.apache.kafka.common.errors.SerializationException;
  5. import org.apache.kafka.common.serialization.Serializer;
  6.  
  7. import java.util.Map;
  8.  
  9. //序列化实现
  10.  
  11. public class GenericSerializer<T> implements Serializer<T> {
  12.  
  13.     private ObjectMapper objectMapper = new ObjectMapper();
  14.  
  15.   // public GenericSerializer(Class<T> pojoClass) {
  16.     public GenericSerializer() {
  17. 	}
  18.  
  19. 	@Override
  20.     public void configure(Map<String, ?> props, boolean isKey) {
  21.     }
  22.  
  23.     @Override
  24.     public byte[] serialize(String topic, T data) {
  25.         if (data == null)
  26.             return null;
  27.  
  28.         try {
  29.             return objectMapper.writeValueAsBytes(data);
  30.         } catch (Exception e) {
  31.             throw new SerializationException("Error serializing JSON message", e);
  32.         }
  33.     }
  34.  
  35.     @Override
  36.     public void close() {
  37.     }
  38.  
  39.  
  40. }
  41.

SerdesFactory.java

 

  1. package cloud.stream.serdes;
  2.  
  3. import java.util.HashMap;
  4. import java.util.Map;
  5.  
  6. import org.apache.kafka.common.serialization.Deserializer;
  7. import org.apache.kafka.common.serialization.Serde;
  8. import org.apache.kafka.common.serialization.Serdes;
  9. import org.apache.kafka.common.serialization.Serializer;
  10.  
  11. import cloud.stream.model.Statistics;
  12.  
  13. public class SerdesFactory {
  14. 	 /**
  15.      * @param <T> The class should have a constructor without any
  16.      *        arguments and have setter and getter for every member variable
  17.      * @param pojoClass POJO class.
  18.      * @return Instance of {@link Serde}
  19.      * 
  20.      * 序列化和反序列化能用方法，
  21.      */
  22.     public static <T> Serde<T> serdFrom(Class<T> pojoClass) {
  23.     	   Map<String, Object> serdeProps = new HashMap<>();
  24.     	 final Serializer<Statistics> statisticsSerializer = new GenericSerializer<>();
  25.          serdeProps.put("JsonPOJOClass", pojoClass);
  26.          statisticsSerializer.configure(serdeProps, false);
  27.  
  28.          final Deserializer<Statistics> statisticsDeserializer = new GenericDeserializer<>();
  29.          serdeProps.put("JsonPOJOClass", pojoClass);
  30.          statisticsDeserializer.configure(serdeProps, false);
  31.  
  32.     	 // return Serdes.serdeFrom(new GenericSerializer<T>(pojoClass), new GenericDeserializer<T>(pojoClass));
  33.     	 return (Serde<T>) Serdes.serdeFrom(statisticsSerializer, statisticsDeserializer);
  34.     }
  35. }

下面是主程序代码

 

  1.  
  2.  * 统计60秒内，温度值的最大值  topic中的消息格式为数字，30， 21或者{"temp":19, "humidity": 25}
  3.  */
  4. public class TemperatureAvgDemo {
  5. 	 private static final int TEMPERATURE_WINDOW_SIZE = 60;
  6.  
  7. 	    public static void main(String[] args) throws Exception {
  8.  
  9. 	        Properties props = new Properties();
  10. 	        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-temp-avg");
  11. 	        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "itcast:9092");
  12. 	        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  13. 	        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  14.  
  15. 	        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  16. 	        props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
  17. 	        
  18. 	        //序列化和反序列化
  19. 	        final Serde<Statistics> statisticsSerde =SerdesFactory.serdFrom( Statistics.class);
  20.  
  21. 	        StreamsBuilder builder = new StreamsBuilder();
  22.  
  23. 	        KStream<String, String> source = builder.stream("snmp-temp1");
  24. 	        KTable<Windowed<String>, Statistics> max = source
  25. 	                .selectKey(new KeyValueMapper<String, String, String>() {
  26. 	                    @Override
  27. 	                    public String apply(String key, String value) {
  28. 	                        return "stat";
  29. 	                    }
  30. 	                })
  31. 	                .groupByKey()
  32. 	                .windowedBy(TimeWindows.of(TimeUnit.SECONDS.toMillis(TEMPERATURE_WINDOW_SIZE)))
  33. 	                .aggregate(
  34. 	                        new Initializer<Statistics>() {
  35. 	                            @Override
  36. 	                            public Statistics apply() {
  37. 	                                Statistics avgAndSum = new Statistics(0L,0L,0L);
  38. 	                                return avgAndSum;
  39. 	                            }
  40. 	                        },
  41. 	                        new Aggregator<String, String, Statistics>() {
  42. 	                            @Override
  43. 	                            public Statistics apply(String aggKey, String newValue, Statistics aggValue) {
  44. 	                                //topic中的消息格式为{"temp":19, "humidity": 25}
  45. 	                                System.out.println("aggKey:" + aggKey + ",  newValue:" + newValue + ", aggKey:" + aggValue);
  46. 	                                Long newValueLong = null;
  47. 	                                try {
  48. 	                                    JSONObject json = JSON.parseObject(newValue);
  49. 	                                    newValueLong = json.getLong("temp");
  50. 	                                }
  51. 	                                catch (ClassCastException ex) {
  52. 	                                     newValueLong = Long.valueOf(newValue);
  53. 	                                }
  54.  
  55. 	                                aggValue.setCount(aggValue.getCount() + 1);
  56. 	                                aggValue.setSum(aggValue.getSum() + newValueLong);
  57. 	                                aggValue.setAvg(aggValue.getSum() / aggValue.getCount());
  58.  
  59. 	                                return aggValue;
  60. 	                            }
  61. 	                        },
  62. 	                        Materialized.<String, Statistics, WindowStore<Bytes, byte[]>>as("time-windowed-aggregated-temp-stream-store")
  63. 	                                .withValueSerde(statisticsSerde)
  64. 	                );
  65.  
  66. 	        WindowedSerializer<String> windowedSerializer = new WindowedSerializer<>(Serdes.String().serializer());
  67. 	        WindowedDeserializer<String> windowedDeserializer = new WindowedDeserializer<>(Serdes.String().deserializer(), TEMPERATURE_WINDOW_SIZE);
  68. 	        Serde<Windowed<String>> windowedSerde = Serdes.serdeFrom(windowedSerializer, windowedDeserializer);
  70. 	        max.toStream().to("reulst-temp-stat", Produced.with(windowedSerde, statisticsSerde));
  71.  
  72. 	        final KafkaStreams streams = new KafkaStreams(builder.build(), props);
  73. 	        final CountDownLatch latch = new CountDownLatch(1);
  74.  
  75.  
  76. 	        Runtime.getRuntime().addShutdownHook(new Thread("streams-temperature-shutdown-hook") {
  77. 	            @Override
  78. 	            public void run() {
  79. 	                streams.close();
  80. 	                latch.countDown();
  81. 	            }
  82. 	        });
  83.  
  84. 	        try {
  85. 	            streams.start();
  86. 	            latch.await();
  87. 	        } catch (Throwable e) {
  88. 	            System.exit(1);
  89. 	        }
  90. 	        System.exit(0);
  91. 	    }
  92. 	
  93. }

最终输出结果

结果输出情况

 

 
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/749815
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号