如何在Java中迭代DStream

Question

我是星火编程的新手。我有一个火花流程序，它需要将接收到的DStream存储到数据库中，我想迭代我的数据流并将每条记录存储到数据库中。

就像这样。

JavaStreamingContext streamingContext = getSparkStreamingContext();

JavaReceiverInputDStream<String> socketTextStream = streamingContext
                .socketTextStream("localhost", 8080);

DStream<String> dstream = socketTextStream.dstream();

// Iterate each record from the DStream and push it to DB

方法2 :

这样做对吗？这种方法会带来任何性能增益/问题吗？

socketTextStream.foreachRDD(new Function<JavaRDD<String>, Void>() {
    @Override
    public Void call(JavaRDD<String> rdd) throws Exception {

        List<String> collect = rdd.collect();

        for (String string : collect) {

            System.out.println(string);
        }

        return null;
    }
});

Answer 1

您可以使用JavaDStream.foreachRDD和JavaRDD.foreach

JavaStreamingContext streamingContext = getSparkStreamingContext();
JavaReceiverInputDStream<String> socketTextStream = streamingContext
        .socketTextStream("localhost", 8080);

socketTextStream.foreachRDD(new VoidFunction<JavaRDD<String>>() {
    @Override
    public void call(JavaRDD<String> rdd) throws Exception {
        rdd.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                // Save data
            }
        });
    }
});

或使用Java 8 Lambda表达式

JavaStreamingContext streamingContext = getSparkStreamingContext();
JavaReceiverInputDStream<String> socketTextStream = streamingContext
        .socketTextStream("localhost", 8080);

socketTextStream.foreachRDD((VoidFunction<JavaRDD<String>>) rdd -> {
    rdd.foreach((VoidFunction<String>) s -> {
        // Save data
    });
});

编辑

既然您使用的是Spark1.2.0(这有点老了，我建议升级一下(最新版本是1.6.1，截止到2016年5月22日))：

socketTextStream.foreachRDD(new Function<JavaRDD<String>, Void>() {
    @Override
    public Void call(JavaRDD<String> rdd) throws Exception {
        rdd.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                // Save data
            }
        });
        return null;
    }
});