apache-spark - 为什么 Spark 流很慢？

Question

我使用了来自 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();
}
}

Answer 1

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

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