Поток искры прекращается - PullRequest
Новая
       38

Поток искры прекращается

0 голосов
/ 14 февраля 2020

Я пытаюсь читать сообщения от Spark Kafka Streaming. Но он останавливается с ошибкой ниже 

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]

1 Ответ

0 голосов
/ 14 февраля 2020

Вы никогда не запускали поток, вызывая для него действие

Набор данных и все преобразования лениво оцениваются.

Вам необходимо распечатать набор данных на терминал или записать его в какую-либо базу данных или hdf, и ds1.col("value") покажет вам несколько строк одновременно, что, вероятно, не то, что вы хотите

Относительно ошибка, у вас нет агрегации, как говорит ошибка. Попробуйте добавить режим вывода

...