gpt4 book ai didi

java - Spark 流停止

转载 作者:行者123 更新时间:2023-12-01 18:26:36 32 4
gpt4 key购买 nike

我正在尝试从 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 从关闭 Hook 调用停止。在新的 Spark 中,它被 sparkSession 覆盖。那为什么现在才来到这里。

friend 们,我可以知道为什么我的直播被停止了吗?理想情况下它应该连续运行。

请大家指教。谢谢..!

更新:

更新代码后出错:

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]

最佳答案

您从未通过调用操作来启动流

数据集和所有转换都是延迟计算的。

您需要将数据集打印到终端或将其写入某个数据库或 hdfs,并且 ds1.col("value") 一次显示多行,这不是您想要的大概想要

关于该错误,正如错误所述,您没有聚合。尝试追加输出模式

关于java - Spark 流停止,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/60219323/

32 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com