将JavaDStream<String>转换为JavaRDD<String>

Question

我有一个JavaDStream，从外部数据源获取数据。我试图集成Spark Streaming和SparkSQL。众所周知，JavaDStream由JavaRDD组成。我只能在拥有JavaRDD时应用applySchema（）函数。请帮我将其转换为JavaRDD。我知道Scala中有函数，而且更容易。但请在Java中帮助我。

- Navin Ahmed

Answer 1

您不能将DStream转换为RDD。正如您提到的，DStream包含RDD。获取对RDD的访问方式是通过对DStream的每个RDD应用函数，使用foreachRDD。请参阅文档：https://spark.apache.org/docs/1.1.0/api/java/org/apache/spark/streaming/api/java/JavaDStreamLike.html#foreachRDD(org.apache.spark.api.java.function.Function2)。

- maasg

---

谢谢。帮了我很多！ - Navin Ahmed

Answer 2

你需要首先使用forEachRDD访问DStream中的所有RDD，如下所示：

javaDStream.foreachRDD( rdd => {
    rdd.collect.foreach({
        ...
    })
})

- amey91

Answer 3

希望这能帮助将JavaDstream转换为JavaRDD！

    JavaDStream<String> lines = stream.map(ConsumerRecord::value);

    //Create JavaRDD<Row>
    lines.foreachRDD(new VoidFunction<JavaRDD<String>>() {
        @Override
        public void call(JavaRDD<String> rdd) {
            JavaRDD<Row> rowRDD = rdd.map(new Function<String, Row>() {
                @Override
                public Row call(String msg) {
                    Row row = RowFactory.create(msg);
                    return row;
                }
            });
            //Create Schema
            StructType schema = DataTypes.createStructType(new StructField[] {
                    DataTypes.createStructField("value", DataTypes.StringType, true)});
            //Get Spark 2.0 session
            SparkSession spark = JavaSparkSessionSingleton.getInstance(rdd.context().getConf());
            Dataset msgDataFrame = spark.createDataFrame(rowRDD, schema);
            msgDataFrame.show();

- AP-Big Data