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

kafkaStream_kstream.map((key,value)->{})

作者：知新_RL | 2024-06-23 14:05:25

赞

踩

kstream.map((key,value)->{})

UserFriendsStream

  1. import org.apache.kafka.clients.consumer.ConsumerConfig;
  2. import org.apache.kafka.common.serialization.Serdes;
  3. import org.apache.kafka.streams.*;
  4. import org.apache.kafka.streams.errors.StreamsException;
  5. import org.apache.kafka.streams.kstream.KStream;
  6.  
  7. import java.util.ArrayList;
  8. import java.util.Properties;
  9. import java.util.concurrent.CountDownLatch;
  10.  
  11. /**
  12.  * 2021.12.28
  13.  * user , friends           topic:user_friends_row
  14.  * 3197468391,1346449342 387324416 4226080662
  15.  *
  16.  * user , friends           topic:user_friends
  17.  * 3197468391 1346449342
  18.  * 3197468391 387324416
  19.  * 3197468391 4226080662
  20.  *
  21.  */
  22. public class UserFriendsStream {
  23.     public static void main(String[] args) {
  24.         Properties prop = new Properties();
  25.         prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"userFriend");
  26.         prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.111.131:9092");
  27.         prop.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,3000);
  28.         prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
  29.         //earliest latest none
  30.         prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
  31.         prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  32.         prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
  33.  
  34.  
  35.         StreamsBuilder builder = new StreamsBuilder();
  36.         KStream<String, String> user_friends_rows = builder.stream("user_friends_rows")
  37.                 .flatMap((key, value) -> { //3197468391,1346449342 387324416 4226080662
  38.                     ArrayList<KeyValue<String, String>> list = new ArrayList<>();
  39.                     String[] fields = value.toString().split(","); //[3197468391,1346449342 387324416 4226080662]
  40.                     if (fields.length == 2) {
  41.                         String[] friends = fields[1].split("\\s+");  //[1346449342 387324416 4226080662]
  42.                         String user = fields[0];                           // 3197468391
  43.                         if (user.trim().length() > 0) {  //  \\s匹配任意空白字符
  44.                             for (String friend :
  45.                                     friends) {
  46.                                 System.out.println(user + "," + friend);
  47.                                 KeyValue<String, String> keyValue = new KeyValue<>(null, user + "," + friend);
  48.                                 list.add(keyValue);
  49.  
  50.                             }
  51.                         }
  52.                     }
  53.                     return list;
  54.  
  55.  
  56.                 });
  57.  
  58.         user_friends_rows.to("user_friends");
  59.  
  60.         //构建 Topology
  61.         Topology topo = builder.build();
  62.         final KafkaStreams streams = new KafkaStreams(topo, prop);
  63.  
  64.         final CountDownLatch latch = new CountDownLatch(1);
  65.  
  66.         Runtime.getRuntime().addShutdownHook(new Thread("stream"){
  67.             @Override
  68.             public void run() {
  69.                 streams.close();
  70.                 latch.countDown();
  71.             }
  72.         });
  73.  
  74.  
  75.  
  76.         try {
  77.             streams.start();
  78.             latch.countDown();
  79.         }catch (IllegalStateException e){
  80.             e.printStackTrace();
  81.         }catch (StreamsException e){
  82.             e.printStackTrace();
  83.     }
  84.  
  85. //        System.exit(0);
  86.  
  87.  
  88.     }
  89. }

 EventAttendStream

  1. import org.apache.kafka.clients.consumer.ConsumerConfig;
  2. import org.apache.kafka.common.serialization.Serdes;
  3. import org.apache.kafka.streams.*;
  4. import org.apache.kafka.streams.errors.StreamsException;
  5. import org.apache.kafka.streams.kstream.KStream;
  6.  
  7. import java.util.ArrayList;
  8. import java.util.Properties;
  9. import java.util.concurrent.CountDownLatch;
  10. //2021.12.28
  11. public class EventAttendStream {
  12.     public static void main(String[] args) {
  13.         Properties prop = new Properties();
  14.         prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"eventattend");
  15.         prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.111.131:9092");
  16.         prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  17.         prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
  18.         prop.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,3000);
  19.         prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false"); //自动提交
  20.         //earliest latest none
  21.         prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
  22.  
  23.  
  24.         StreamsBuilder builder = new StreamsBuilder();
  25.         KStream<Object, Object> ear = builder.stream("event_attendess_row"); //流
  26.         ear.flatMap((key,value)->{
  27.             ArrayList<KeyValue<String,String>>list=new ArrayList<>();
  28.             String[] fields = value.toString().split(",");
  29.             String eventid = fields[0];
  30.             //yes
  31.             if (fields.length>=2 && fields[1].trim().length()>0){
  32.                 String[]yes=fields[1].trim().split("\\s+");
  33.                 for (String y:
  34.                 yes){
  35.                     System.out.println(eventid+","+y+",yes");
  36.                     KeyValue<String, String> yesKeyValue = new KeyValue<>(null, eventid + "," + y + ",yes");
  37.                     list.add(yesKeyValue);
  38.                 }
  39.  
  40.             }
  41.  
  42.             //maybe
  43.             if (fields.length>=3 && fields[2].trim().length()>0){
  44.                 String[]maybe=fields[2].trim().split("\\s+");
  45.                 for (String mb:
  46.                         maybe){
  47.                     System.out.println(eventid+","+mb+",maybe");
  48.                     KeyValue<String, String> mbKeyValue = new KeyValue<>(null, eventid + "," + mb + ",maybe");
  49.                     list.add(mbKeyValue);
  50.                 }
  51.  
  52.             }
  53.  
  54.             //invited
  55.             if (fields.length>=4 && fields[3].trim().length()>0){
  56.                 String[]invited=fields[3].trim().split("\\s+");
  57.                 for (String i:
  58.                         invited){
  59.                     System.out.println(eventid+","+i+",invited");
  60.                     KeyValue<String, String> invitedKeyValue = new KeyValue<>(null, eventid + "," + i + ",invited");
  61.                     list.add(invitedKeyValue);
  62.                 }
  63.  
  64.             }
  65.             //no
  66.             if (fields.length>=5 && fields[4].trim().length()>0){
  67.                 String[]nos=fields[4].trim().split("\\s+");
  68.                 for (String no:
  69.                         nos){
  70.                     System.out.println(eventid+","+no+",no");
  71.                     KeyValue<String, String> noKeyValue = new KeyValue<>(null, eventid + "," + no + ",no");
  72.                     list.add(noKeyValue);
  73.                 }
  74.  
  75.             }
  76.  
  77.  
  78.  
  79.             return list;
  80.  
  81.  
  82.         }).to("event_attendess");
  83.  
  84.         Topology topo = builder.build();
  85.         final KafkaStreams streams = new KafkaStreams(topo, prop);
  86.         final CountDownLatch latch = new CountDownLatch(1);
  87.  
  88.         Runtime.getRuntime().addShutdownHook(new Thread("stream"){
  89.             @Override
  90.             public void run() {
  91.                 streams.close();
  92.                 latch.countDown();
  93.             }
  94.         });
  95.  
  96.         try {
  97.             streams.start();
  98.             latch.await();
  99.         } catch (IllegalStateException e) {
  100.                 e.printStackTrace();
  101.         } catch (StreamsException e) {
  102.             e.printStackTrace();
  103.         } catch (InterruptedException e) {
  104.             e.printStackTrace();
  105.         }
  106.  
  107.         System.exit(0);
  108.  
  109.  
  110.     }
  111.  
  112. }

 MyStreamTest

  1. import org.apache.kafka.clients.consumer.ConsumerConfig;
  2. import org.apache.kafka.common.serialization.Serdes;
  3. import org.apache.kafka.streams.KafkaStreams;
  4. import org.apache.kafka.streams.StreamsBuilder;
  5. import org.apache.kafka.streams.StreamsConfig;
  6. import org.apache.kafka.streams.Topology;
  7. import org.apache.kafka.streams.errors.StreamsException;
  8. import org.apache.kafka.streams.kstream.KStream;
  9.  
  10. import java.util.Properties;
  11. import java.util.concurrent.CountDownLatch;
  12.  
  13. public class MyStreamTest {
  14.     public static void main(String[] args) {
  15.         Properties prop = new Properties();
  16.         prop.put(StreamsConfig.APPLICATION_ID_CONFIG, "myStream");
  17.         prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.111.131:9092");
  18.         prop.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 3000);
  19.         prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
  20.         //earliest latest none
  21.         prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  22.         prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  23.         prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  24.  
  25.         StreamsBuilder builder = new StreamsBuilder();
  26.         KStream<Object, Object> in = builder.stream("mystreamin");
  27.         in.to("mystreamout");
  28.  
  29.  
  30.         Topology topo = builder.build();
  31.         final KafkaStreams streams = new KafkaStreams(topo, prop);
  32.         final CountDownLatch latch = new CountDownLatch(1);
  33.  
  34.         Runtime.getRuntime().addShutdownHook(new Thread("stream"){
  35.             @Override
  36.             public void run() {
  37.                 streams.close();
  38.                 latch.countDown();
  39.             }
  40.         });
  41.  
  42.         try {
  43.             streams.start();
  44.             latch.await();
  45.         } catch (IllegalStateException e) {
  46.             e.printStackTrace();
  47.         } catch (StreamsException e) {
  48.             e.printStackTrace();
  49.         } catch (InterruptedException e) {
  50.             e.printStackTrace();
  51.         }
  52.  
  53.         System.exit(0);
  54.  
  55.  
  56.     }
  57.  
  58. }
  59.  
  60.  
  61.

本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/749784
推荐阅读