为什么火花流很慢？答案

【问题标题】：Why spark streaming is slow?为什么火花流很慢？
【发布时间】：2015-09-10 23:26:34
【问题描述】：

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

【问题讨论】：

这个问题需要详细一点，有多少数据，你在做什么操作。另外仅供参考，在单个节点上运行 spark 基本上没有意义
我只是在测试火花流。我使用了wordcount程序。因为想先学后学，创建集群，在集群上运行主要的复杂任务。
你的窗口大小是多少，如果大于 20 秒，结果会在 20-30 秒后开始出现，然后希望之后更快，你也实现了聚合和反向聚合字数统计功能
我没有定义窗口大小。什么是聚合和逆聚合？
我认为如果您实际发布了您正在运行的代码，这将更容易回答

标签： apache-spark spark-streaming

【解决方案1】：

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

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

【讨论】：

【解决方案2】：

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

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

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

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

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

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

【讨论】：