java上的spark流Dataframe和累加器

cld4siwp 于 2021-05-29 发布在 Spark

关注(0)|答案(1)|浏览(583)

我正在spark structured streaming中处理kafka json流。作为微批处理，我可以使用流式Dataframe的累加器吗？

LongAccumulator longAccum = new LongAccumulator("my accum");

Dataset<Row> df2 = df.filter(output.col("Called number").equalTo("0860"))
            .groupBy("Calling number").count();
// put row counter to accumulator for example
df2.javaRDD().foreach(row -> {longAccumulator.add(1);})

投掷

Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;

. 我也很困惑这样使用蓄能器。将Dataframe向下转换为rdd看起来很奇怪，而且不必要。我可以不用c rdd和foreach（）吗？
根据exeption，我从源dataframe中删除了foreach，并在writestream（）中完成了它

StreamingQuery ds = df2
            .writeStream().foreachBatch( (rowDataset, aLong) -> {
                longAccum.add(1);
                log.info("accum : " + longAccum.value());
            })
            .outputMode("complete")
            .format("console").start();

它正在工作，但我在日志中没有值，在gui中也看不到累加器。

Java streaming apache-spark Accumulator

来源：https://stackoverflow.com/questions/62424833/spark-streaming-dataframes-and-accumulators-on-java

1条答案

按热度按时间

mm5n2pyu1#

不，您可以使用下面的数据集直接访问-

LongAccumulator longAccum = spark.sparkContext().longAccumulator("my accum");

        Dataset<Row> df = spark.range(100).withColumn("x", lit("x"));

        //access in map
        df.map((MapFunction<Row, Row>) row -> {
            longAccum.add(1);
            return  row;
        }, RowEncoder.apply(df.schema()))
                .count();

        // accumulator value
        System.out.println(longAccum.value()); // 100

        longAccum.reset();
        // access in for each
        df.foreach((ForeachFunction<Row>) row -> longAccum.add(1));

        // accumulator value
        System.out.println(longAccum.value()); // 100

请注意，累加器值仅在 action 执行。

使用流式Dataframe

longAccum.reset();
        /**
         * streaming dataframe from csv dir
         * test.csv
         * --------
         * csv
         * id,name
         * 1,bob
         * 2,smith
         * 3,jam
         * 4,dwayne
         * 5,mike
         */
        String fileDir = getClass().getResource("/" + "csv").getPath();
        StructType schema = new StructType()
                .add(new StructField("id", DataTypes.LongType, true, Metadata.empty()))
                .add(new StructField("name", DataTypes.StringType, true, Metadata.empty()));
        Dataset<Row> json = spark.readStream().schema(schema).option("header", true).csv(fileDir);

        StreamingQuery streamingQuery = json
                .map((MapFunction<Row, Row>) row -> {
                    longAccum.add(1);
                    return row;
                }, RowEncoder.apply(df.schema()))
                .writeStream()
                .format("console").start();
        streamingQuery.processAllAvailable();

        // accumulator value
        System.out.println(longAccum.value()); // 5

赞(0）回复(0）举报 2021-05-29

我来回答

java上的spark流Dataframe和累加器

1条答案

使用流式Dataframe

相关问题

热门标签

最新问答