Spark Streaming writeStream未写入文件

文章/答案/技术大牛

发布

1回答

scala、apache-spark

如果我的输出格式是"console"，那么writeStream似乎可以工作，但当我的输出格式是"parquet“时，它就不起作用了。我是否遗漏了一些额外的设置或配置？import org.apache.spark.sql.DataFrame )) def readFromParquet()

浏览 19提问于2021-06-02得票数 0

回答已采纳

1回答

结构化流式处理spark.sql.streaming.schemaInference不处理架构更改

apache-spark、spark-streaming

sparkSession.config("spark.sql.streaming.schemaInference", true).getOrCreate();dataset.writeStreamfile:/save-parquet-files

浏览 2提问于2021-07-09得票数 0

3回答

如何使用Azure宇宙DB连接器写入CosmosDB？

scala、apache-spark、azure-cosmosdb、spark-structured-streaming

当我调用writeStream方法时，下面会弹出错误。添加到集群中的库版本是：我的代码如下： .writeStream .options(config).start这将导致错误：命令-7516

浏览 8提问于2019-11-20得票数 1

1回答

Spark Job未将消息发布到Kafka主题

scala、apache-spark、apache-kafka、spark-streaming、kafka-producer-api

我已经编写了spark作业来读取一个文件，将数据转换为json并将数据发布到Kafka:我尝试了所有选项，如1. posting thread.sleep 2.更改linger.ms小于thread.sleep.But以下是我的项目的重要文件：名称:= "SparkStreamingExample“scalaVersion := "2.11.8" val spark" % "spark

浏览 0提问于2020-06-10得票数 0

1回答

结构化流式传输性能和清除地块文件

apache-spark、parquet、spark-structured-streaming

我正在使用Spark structured streaming从Kafka获取流数据。我需要聚合各种指标(比方说6个指标)，并编写为拼图文件。我确实看到指标1和指标2之间存在巨大的延迟。另外，我还编写了另一个应用程序应该读取的拼图文件。如何不断清除旧的实木地板信息？我应该有一个不同的应用程序吗？Dataset<String> lines_topic = spark.readStream().format("kafka").option("kafka.bootstr

浏览 3提问于2018-01-06得票数 1

回答已采纳

1回答

Apache Spark* Structured Streaming (DataStreamWriter)写入配置单元表*

apache-spark、apache-kafka、apache-spark-sql

我希望使用火花结构流从卡夫卡读取数据，并处理它，并写入蜂窝表。val spark = SparkSession .appName("Kafka Test") .config("spark.streaming.backpressure.enabled", "true"

浏览 5提问于2018-02-06得票数 1

2回答

结构化流传输将Parquet文件写入Hadoop

apache-spark、spark-structured-streaming

我能够将结构化流式传输的结果写入到拼图文件中。问题是这些文件在本地文件系统中，现在我想将它们写入Hadoop文件系统。有没有办法做到这一点？$.read(StreamMetadata.scala:51)at org.apache.spark</e

浏览 9提问于2017-03-01得票数 1

回答已采纳

2回答

结构化流式写入到不同的拼图文件夹

apache-spark、apache-spark-sql、parquet、spark-structured-streaming

我正在使用spark structured streaming从kafka主题中读取事件，并对其进行处理并将其写入到parquet中。我必须根据我在事件中获得的键将输出写入不同的文件夹。我尝试使用结构化流式处理示例始终指向特定的文件夹.Do我需要为每个文件夹启动一个流吗？df.writeStream.format("parquet").option("path", "path/to/

浏览 2提问于2018-08-13得票数 0

1回答

如何访问后续批处理查询的“内存”表中的流查询数据？

scala、apache-spark、spark-structured-streaming

给定一个writeStream调用： .outputMode(outputMode) .queryNameval df = spark.sql("""select Origin,Dest,Carrier,avg(DepDelay) avgDepDelay from MyInMemoryTablegroup by 1,2,3""&quo

浏览 0提问于2019-07-16得票数 2

回答已采纳

1回答

在Spark* Structured Streaming中使用Kafka接收器时，检查点是强制的吗？*

java、apache-spark、spark-structured-streaming

我正在尝试使用Spark Structured Streaming将聚合数据写入Kafka。下面是我的代码： .writeStream() .outputMode(OutputMode.Append())", ...); at org.apache.spark.sql.streaming.StreamingQueryManager$$anonfun$3.apply(StreamingQu

浏览 1提问于2018-04-10得票数 1

2回答

数据源io.pivotal.greenplum.spark.GreenplumRelationProvider不支持流写入。

scala、apache-kafka、spark-streaming、greenplum

我使用的是格林梅-火花连接器，但我正在获取数据源，io.pivotal.greenplum.spark.GreenplumRelationProvider不支持流写入。是否格林梅源不支持流媒体数据？我曾尝试将数据源命名为“绿梅”，并将"io.pivotal.greenplum.spark.GreenplumRelationProvider“转换为.format(”数据源“) .format("kafka") .opti

浏览 0提问于2019-04-04得票数 0

回答已采纳

2回答

无法使用Spark结构化流在Parquet文件中写入数据

scala、apache-spark、spark-structured-streaming

我有一个星火结构的流： .readStream .option("kafka.bootstrap.serversoption("endingOffsets", "latest") .load()

浏览 5提问于2017-05-25得票数 4

回答已采纳

1回答

Spark Structured Streaming应用程序将空拼图文件生成到Azure blob

scala、spark-streaming

我从Apache Kafka读取json消息，然后使用Apache Spark将拼图文件写入Azure blob存储中。我使用方法partitionBy将这些拼图文件写入嵌套文件夹中。我的代码是这样的： .format("kafka") .load()

浏览 1提问于2020-10-26得票数 2

2回答

为什么我的查询因AnalysisException而失败？

scala、apache-spark、spark-structured-streaming

我是Spark streaming的新手。我正在尝试使用本地csv文件进行结构化Spark流。我在处理的时候遇到了下面的异常。Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources mustbe executed with writeStream.start();; FileSource[file:/&

浏览 0提问于2017-07-28得票数 0

1回答

如何让spark同时运行一个作业中的所有任务？

python、apache-spark、jvm、pyspark、bigdata

我有一个系统，其中REST API (Flask)使用spark-sumbit将作业发送到启动并运行的pyspark。由于各种原因，我需要spark来同时运行所有任务(例如，我需要设置executors的数量=运行时的任务数量)。例如，如果我有20个任务，但只有4个核心，我希望每个核心执行5个任务(执行器)，而不必重新启动spark。我知道我可以在启动spark时设置executors的数量，但我不想这样做，因为spark正在执行其他作业。这可以通过一种变通方法来实现吗？

浏览 0提问于2017-08-26得票数 0

3回答

如何将火花结构流DataFrame插入到Hive外部表/位置？

apache-spark、hive、spark-structured-streaming

这是我的例子 .enableHiveSupport() .config("hive.exec.dynamic.partition.mode", "nonstrict") .config("spark.sql.<em

浏览 0提问于2018-12-28得票数 13

1回答

如何从Kafka访问记录中的文件路径并从创建数据集？

java、apache-spark、apache-kafka、spark-structured-streaming

我正在接收Kafka消息上的文件路径。我需要将这个文件加载到spark RDD中，对其进行处理，然后将其转储到HDFS。Queries with streaming sources must be executed with writeStream.start();" Queries with streaming so

浏览 0提问于2017-09-26得票数 4

1回答

Spark Structured Streaming:未指定路径错误

apache-spark、apache-kafka、spark-structured-streaming

尝试使用Spark Structured Streaming将数据写入Kafka主题时出现以下错误。aggregatedDataset .writeStream() at org.apache.spark.sql.execut

浏览 1提问于2018-09-28得票数 0

1回答

如何编写hive格式的结构化查询结果？

scala、apache-spark、hive、spark-structured-streaming

rdf.writeStream.format("hive").partitionBy("date") .option("db", "default") .queryName("socket-hive-<e

浏览 29提问于2018-11-29得票数 1

2回答

如何使用FileFormat格式的更新输出模式？

scala、apache-spark、spark-structured-streaming

我正在尝试在更新输出模式下使用spark structured写入文件。我找到了，只要配置了控制台格式，它就能正常工作。但如果我将输出模式更改为： .writeStream .format("json"): Data source json does not support Update output mode; at org.apache.spar

浏览 12提问于2018-02-28得票数 3

点击加载更多