火花流停止

Question

我正在尝试从 Spark kafka 流中读取消息。但它因以下错误而停止

20/02/14 08:12:33 INFO SparkContext: Invoking stop() from shutdown hook
20/02/14 08:12:33 INFO SparkUI: Stopped Spark web UI at http://192.000.0.777:3333
20/02/14 08:12:33 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
20/02/14 08:12:33 INFO MemoryStore: MemoryStore cleared
20/02/14 08:12:33 INFO BlockManager: BlockManager stopped
20/02/14 08:12:33 INFO BlockManagerMaster: BlockManagerMaster stopped
20/02/14 08:12:33 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
20/02/14 08:12:33 INFO SparkContext: Successfully stopped SparkContext
20/02/14 08:12:33 INFO ShutdownHookManager: Shutdown hook called
20/02/14 08:12:33 INFO ShutdownHookManager: Deleting directory /tmp/spark-34e4907e-cc7f-4630
20/02/14 08:12:33 INFO ShutdownHookManager: Deleting directory /tmp/temporaryReader-63fe1c85-68b8-4906

这是我的代码

package sparkProject;

import static org.apache.spark.sql.functions.callUDF;

import java.util.*;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.types.*;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;

public class XMLSparkStreamEntry {

    static {
        StructType customSchema = new StructType(
                new StructField[] { new StructField("id", DataTypes.StringType, true, Metadata.empty()),
                        new StructField("author", DataTypes.StringType, true, Metadata.empty()),
                        new StructField("description", DataTypes.StringType, true, Metadata.empty()),
                        new StructField("genre", DataTypes.StringType, true, Metadata.empty()),
                        new StructField("price", DataTypes.DoubleType, true, Metadata.empty()),
                        new StructField("publish_date", DataTypes.StringType, true, Metadata.empty()),
                        new StructField("title", DataTypes.StringType, true, Metadata.empty()) });

    }

    public static void registerPrintValue(SparkSession spark) {

        spark.udf().register("registerPrintValue", new UDF1<String, String>() {

            /**
             * 
             */
            private static final long serialVersionUID = 1L;

            @Override
            public String call(String t1) throws Exception {

                System.out.println("Value: " + t1);
                return t1;
            }

        }, DataTypes.StringType);

    }

    public static void main(String[] args) throws StreamingQueryException {

        SparkConf conf = new SparkConf();
        SparkSession spark = SparkSession.builder().config(conf).appName("Spark Program").master("local[*]")
                .getOrCreate();

        Dataset<Row> ds1 = spark.readStream().format("kafka").option("kafka.bootstrap.servers", "localhost:9092")
                .option("subscribe", "Kafkademo").load();

        Dataset<Row> stringTypeDS = ds1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)");
        System.out.println("Key:" + ds1.col("value").toString());
        XMLSparkStreamEntry.registerPrintValue(spark);
        callUDF("registerPrintValue", stringTypeDS.col("value"));

        StreamingQuery query = stringTypeDS.writeStream().outputMode("complete").format("console").start();

        query.awaitTermination();

    }

}

我正在使用 Spark 2.4 最新版本的 spark。那么为什么它显示错误SparkContext 调用停止从关闭挂钩。在新的火花中，它被sparkSession 覆盖。那为什么它现在来这里。

我可以知道朋友们为什么我的流媒体被停止了吗？理想情况下，它应该连续运行。

请插嘴。 谢谢..！

更新：

更新代码后出错：

Exception in thread "main" org.apache.spark.sql.AnalysisException: Complete output mode not supported when there are no streaming aggregations on streaming DataFrames/Datasets;;
Project [cast(key#7 as string) AS key#21, cast(value#8 as string) AS value#22]
+- StreamingRelationV2 org.apache.spark.sql.kafka010.KafkaSourceProvider@3cc053, kafka, Map(subscribe -> Kafkademo, kafka.bootstrap.servers -> localhost:9092), [key#7, value#8, topic#9, partition#10, offset#11L, timestamp#12, timestampType#13], StreamingRelation DataSource(org.apache.spark.sql.SparkSession@7fdd43cd,kafka,List(),None,List(),None,Map(subscribe -> Kafkademo, kafka.bootstrap.servers -> localhost:9092),None), kafka, [key#0, value#1, topic#2, partition#3, offset#4L, timestamp#5, timestampType#6]

Answer 1

你从来没有通过调用动作来启动流

数据集和所有转换都是延迟评估的。

您需要将 Dataset 打印到终端或将其写入某些数据库或 hdfs，ds1.col("value") 一次向您显示多行，这可能不是您想要的

关于错误，正如错误所说，您没有聚合。试试追加输出模式