java - 转换 JavaDStream到 JavaRDD

Question

我有一个从外部源获取数据的 JavaDStream。我正在尝试集成 Spark Streaming 和 SparkSQL。众所周知，JavaDStream 是由 JavaRDD 的 . 当我有 JavaRDD 时，我只能应用函数 applySchema()。请帮助我将其转换为 JavaRDD。我知道 scala 中有一些函数，而且它更容易。但是用Java帮助我。

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)

Answer 2

您必须首先使用 forEachRDD 访问 DStream 中的所有 RDD，如下所示：

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

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();