Kafka Spark流媒体：无法阅读消息

2 回答

• 7

  问题解决了 .

  上面的代码是正确的 . 我们将再添加两行来抑制生成的[INFO]和[WARN] . 所以最终的代码是：

  package com.spark;
  
  import scala.Tuple2;
  import org.apache.log4j.Logger;
  import org.apache.log4j.Level;
  import kafka.serializer.Decoder;
  import kafka.serializer.Encoder;
  import org.apache.spark.streaming.Duration;
  import org.apache.spark.*;
  import org.apache.spark.api.java.function.*;
  import org.apache.spark.api.java.*;
  import org.apache.spark.streaming.kafka.KafkaUtils;
  import org.apache.spark.streaming.kafka.*;
  import org.apache.spark.streaming.api.java.JavaStreamingContext;
  import org.apache.spark.streaming.api.java.JavaPairDStream;
  import org.apache.spark.streaming.api.java.JavaDStream;
  import org.apache.spark.streaming.api.java.JavaPairReceiverInputDStream;
  import java.util.Map;
  import java.util.HashMap;
  
  public class SparkStream {
      public static void main(String args[])
      {
          if(args.length != 3)
          {
              System.out.println("SparkStream <zookeeper_ip> <group_nm> <topic1,topic2,...>");
              System.exit(1);
          }
  
          Logger.getLogger("org").setLevel(Level.OFF);
          Logger.getLogger("akka").setLevel(Level.OFF);
          Map<String,Integer> topicMap = new HashMap<String,Integer>();
          String[] topic = args[2].split(",");
          for(String t: topic)
          {
              topicMap.put(t, new Integer(3));
          }
  
          JavaStreamingContext jssc = new JavaStreamingContext("local[4]", "SparkStream", new Duration(1000));
          JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createStream(jssc, args[0], args[1], topicMap );
  
          System.out.println("Connection done++++++++++++++");
          JavaDStream<String> data = messages.map(new Function<Tuple2<String, String>, String>() 
                                                  {
                                                      public String call(Tuple2<String, String> message)
                                                      {
                                                          return message._2();
                                                      }
                                                  }
                                                  );
          data.print();
  
          jssc.start();
          jssc.awaitTermination();
  
      }
  }
  

  我们还需要在POM.xml中添加依赖项：

  <dependency>
  <groupId>com.msiops.footing</groupId>
  <artifactId>footing-tuple</artifactId>
  <version>0.2</version>
  </dependency>
  

  此依赖项用于使用 scala.Tuple2
Stream 0 received 0 block 的错误是由于spark-worker不可用而且spark-worker-core设置为1.对于spark-streaming，我们需要核心> = 2 . 所以我们需要在spark-config文件中进行更改 . 请参阅安装手册 . 添加行 export SPARK_WORKER_CORE=5 也将 SPARK_MASTER='hostname' 更改为 SPARK_MASTER=<your local IP> . 当您访问Spark UI Web控制台时，您在BOLD中看到的本地IP ...类似于： spark://192.168..:<port> . 我们这里不需要这个端口 . 只需要IP .
  现在重新启动你的spark-master和spark-worker并开始流式传输:)

  输出：

  -------------------------------------------
  Time: 1417443060000 ms
  -------------------------------------------
  message 1
  
  -------------------------------------------
  Time: 1417443061000 ms
  -------------------------------------------
  message 2
  
  -------------------------------------------
  Time: 1417443063000 ms
  -------------------------------------------
  message 3
  message 4
  
  -------------------------------------------
  Time: 1417443064000 ms
  -------------------------------------------
  message 5
  message 6
  messag 7
  
  -------------------------------------------
  Time: 1417443065000 ms
  -------------------------------------------
  message 8
  

  回复于 2024-05-01T02:11:45+08:00
• 2

  是的，您需要从DStream访问内容 .

  messages.foreachRDD(<<processing for the input received in the interval>>);
  

  回复于 2024-05-01T02:11:45+08:00