最近在思考kafka Stream 是否可以替换掉flume

电玩女神 2024-04-17 22:45 25阅读 0赞

1，先写了个简单的案例，本地运行成功

package com.coder.flink.core.a_kafkaStream;
    
    
    
    import java.util.Arrays;
    import java.util.Properties;
    import java.util.concurrent.CountDownLatch;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.common.serialization.Serdes;
    import org.apache.kafka.streams.*;
    import org.apache.kafka.streams.errors.StreamsException;
    import org.apache.kafka.streams.kstream.*;
    
    
    public class WordCountApplication {
        public static void main(final String[] args) throws Exception {
            Properties props = new Properties();
    
            //todo 设置名称
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
    
            //todo 设置节点
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "node2.hadoop:9092,node3.hadoop:9092");
    
            //todo 设置线程数
            props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "4");
    
            //todo 设置序列化
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
    
            //todo 指定消费偏移量策略
    
            //
            StreamsBuilder builder = new StreamsBuilder();
    
            //todo 指定topic
            KStream<String, String> sourceData = builder.stream("stream_in");
    //        sourceData.flatMapValues(new ValueMapper<String, Iterable<String>>() {
    //            @Override
    //            public Iterable<String> apply(String s) {
    //                return null;
    //            }
    //        });
    
    
    
            //todo 创建table
            KStream<String, String> wordCounts = sourceData
                    .map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                        @Override
                        public KeyValue<String, String> apply(String s, String s2) {
                            System.out.println("s = " + s);
                            return new KeyValue<>(s, s2);
                        }
                    });
    
                   /* .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
    
                    .selectKey((key, word) -> word)
    
                    .groupByKey()
    
                    .count(Materialized.as("counts-store"));*/
    
            //todo 写入到新的topic
            wordCounts.print(Printed.toSysOut());
            wordCounts.to("stream_out");
    
            //todo 创建Topology
            Topology topology = builder.build();
    //        System.out.println(topology.describe());
    
            //todo 创建kafka stream
            KafkaStreams streams = new KafkaStreams(topology, props);
    
           final CountDownLatch latch =  new CountDownLatch(1);
    
    
            Runtime.getRuntime().addShutdownHook(new Thread("aaa"){
                @Override
                public void run() {
                    streams.close();
                    latch.countDown();
                }
            });
    
            //todo 启动
            try {
                streams.start();
                latch.await();
            } catch (IllegalStateException e) {
                e.printStackTrace();
            } catch (StreamsException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
    }

2，kafka stream 自带案例运行：

bin/kafka-run-class.sh org.apache.kafka.streams.examples.wordcount.WordCountDemo

3，需要验证是否可以取代掉flume，需要思考的点

1）是否会丢数据

2）是否有序，吞吐量怎么样

3）资源消耗怎么样

4，好的文章参考：

[https://blog.csdn.net/gangchengzhong/article/details/84023095][https_blog.csdn.net_gangchengzhong_article_details_84023095]

[https_blog.csdn.net_gangchengzhong_article_details_84023095]: https://blog.csdn.net/gangchengzhong/article/details/84023095