开发火花流应用程序

Question

所以我要解决的问题如下：

• 我需要一个以特定频率发出消息的数据源
• 有 N 个神经网络需要单独处理每条消息
• 所有神经网络的输出都被聚合，只有当每个消息的所有 N 个输出都被收集时，才应该声明消息已完全处理
• 最后，我应该测量消息被完全处理所花费的时间（从发出消息到收集该消息的所有 N 个神经网络输出之间的时间）

我很好奇如何使用火花流处理这样的任务。

我当前的实现使用 3 种类型的组件：一个自定义接收器和两个实现 Function 的类，一个用于神经网络，一个用于最终聚合器。

概括地说，我的应用程序是这样构建的：

JavaReceiverInputDStream<...> rndLists = jssc.receiverStream(new JavaRandomReceiver(...));

Function<JavaRDD<...>, Void> aggregator = new JavaSyncBarrier(numberOfNets);

for(int i = 0; i < numberOfNets; i++){
    rndLists.map(new NeuralNetMapper(neuralNetConfig)).foreachRDD(aggregator);
}

不过，我遇到的主要问题是它在本地模式下的运行速度比提交到 4 节点集群时要快。

我的实现一开始是错误的还是这里发生了其他事情？

这里还有一个完整的帖子http://apache-spark-user-list.1001560.n3.nabble.com/Developing-a-spark-streaming-application-td12893.html，其中详细介绍了前面提到的三个组件中的每一个的实现。

Answer 1

似乎有很多重复的对象实例化和序列化。后者可能会影响您在集群中的性能。

您应该只尝试一次实例化您的神经网络。您必须确保它们是可序列化的。您应该使用flatMap 而不是多个maps + union。大致如下：

// Initialize neural net first
List<NeuralNetMapper> neuralNetMappers = new ArrayList<>(numberOfNets);
for(int i = 0; i < numberOfNets; i++){
    neuralNetMappers.add(new NeuralNetMapper(neuralNetConfig));
}

// Then create a DStream applying all of them
JavaDStream<Result> neuralNetResults = rndLists.flatMap(new FlatMapFunction<Item, Result>() {
    @Override
    public Iterable<Result> call(Item item) {
        List<Result> results = new ArrayList<>(numberOfNets);
        for (int i = 0; i < numberOfNets; i++) {
            results.add(neuralNetMappers.get(i).doYourNeuralNetStuff(item));
        }
        return results;
    }
});

// The aggregation stuff
neuralNetResults.foreachRDD(aggregator);

如果您负担得起以这种方式初始化网络，您可以节省大量时间。此外，您在链接帖子中包含的 union 内容似乎是不必要的，并且会影响您的表现：flatMap 可以。

最后，为了进一步调整您在集群中的性能，you can use the Kryo serializer。