4

我使用了来自 github 存储库的 spark 流示例程序，并尝试使用 kafka 和自定义接收器。在这两种情况下，我都会在 20-30 秒后得到输出。在自定义接收器代码中，我立即获取数据，但输出需要 20-30 秒。我在单个节点上运行此代码。

我做错了什么还是有优化，我需要执行还是因为我在单个节点上运行。

如果有人可以指导我这将是一个很大的帮助。

我使用了 spark 存储库中的代码，代码如下：

import scala.Tuple2;
import com.google.common.collect.Lists;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.StorageLevels;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import java.util.regex.Pattern;

/**
 * Counts words in UTF8 encoded, '\n' delimited text received from the network every second.
 * Usage: JavaNetworkWordCount <hostname> <port>
 *   <hostname> and <port> describe the TCP server that Spark Streaming would connect to receive       data.
 *
 * To run this on your local machine, you need to first run a Netcat server
 *    `$ nc -lk 9999`
 * and then run the example
 *    `$ bin/run-example org.apache.spark.examples.streaming.JavaNetworkWordCount localhost 9999`
 */
 public final class JavaNetworkWordCount {
    private static final Pattern SPACE = Pattern.compile(" ");

public static void main(String[] args) {
  if (args.length < 2) {
    System.err.println("Usage: JavaNetworkWordCount <hostname> <port>");
    System.exit(1);
}

StreamingExamples.setStreamingLogLevels();

// Create the context with a 1 second batch size
SparkConf sparkConf = new SparkConf().setAppName("JavaNetworkWordCount");
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf,  new Duration(1000));

// Create a JavaReceiverInputDStream on target ip:port and count the
// words in input stream of \n delimited text (eg. generated by 'nc')
// Note that no duplication in storage level only for running locally.
// Replication necessary in distributed scenario for fault tolerance.
JavaReceiverInputDStream<String> lines = ssc.socketTextStream(
        args[0], Integer.parseInt(args[1]), StorageLevels.MEMORY_AND_DISK_SER);
JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
  @Override
  public Iterable<String> call(String x) {
    return Lists.newArrayList(SPACE.split(x));
  }
});
JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
  new PairFunction<String, String, Integer>() {
    @Override
    public Tuple2<String, Integer> call(String s) {
      return new Tuple2<String, Integer>(s, 1);
    }
  }).reduceByKey(new Function2<Integer, Integer, Integer>() {
    @Override
    public Integer call(Integer i1, Integer i2) {
      return i1 + i2;
    }
  });

wordCounts.print();
ssc.start();
ssc.awaitTermination();
}
}
4

2 回答 2

1

我在任何地方都看不到主配置。当您说您在一个节点上运行时，我猜您的意思是“本地”模式而不是独立的单个节点。如果是这种情况，默认情况下本地使用单个线程，该线程将由接收的执行程序使用并饿死执行程序。

尝试更改 new SparkConf().setAppName("JavaNetworkWordCount"); 到新 SparkConf().setAppName("JavaNetworkWordCount").setMaster("local[4]");

于 2015-05-14T13:54:45.643 回答
0

我想你的意思是local[*]主人，否则程序将无法启动。我也希望你没有设置local[1]，因为这也是本地模式的错误。（Spark 将无法正常执行）。请参阅上面 Gonzalo 的回答。

我还假设您或多或少拥有现代硬件。

我刚刚实际运行了这个简单的示例，并且在通过ncjavadoc 建议的方式发送输入时没有遇到任何延迟或其他问题。

据我所知，这段代码在性能方面没有任何问题：琐碎的操作，没有什么可改进的。从 Java 和 Spark 的角度来看。RDD 缓存在这里没有帮助，因为没有重复计算（RDD 只打印一次）。

您提到了 Kafka，但代码没有引用它。如果它仍然是实际的，请同时发布相关代码。

否则，没有足够的证据称 Spark 流式传输速度慢 :)。

于 2015-09-10T15:26:34.047 回答