Spark streaming from eventhub:一旦没有更多的数据，如何停止流？
Spark Streaming是Apache Spark的一个组件，用于实时处理和分析数据流。EventHub是Azure提供的一种事件处理服务，用于接收和处理大规模实时数据。
在Spark Streaming中，要停止流，可以使用StreamingContext.stop()方法来停止StreamingContext对象。该方法将停止接收新的数据，并在所有已接收的数据处理完成后优雅地关闭流。
以下是一个示例代码，展示了如何停止Spark Streaming流：
from pyspark.streaming import StreamingContext
# 创建StreamingContext对象
ssc = StreamingContext(sparkContext, batchDuration)
# 创建DStream，从EventHub接收数据
dstream = EventHubsUtils.createDirectStreams(ssc, eventHubNamespace, eventHubName, eventHubPolicyName, eventHubPolicyKey)
# 对DStream进行处理
dstream.foreachRDD(processData)
# 启动流
ssc.start()
# 等待流停止
ssc.awaitTermination()
# 停止流
ssc.stop()在上述示例中，ssc.awaitTermination()方法将使程序一直运行，直到手动停止或发生错误。当需要停止流时，可以通过调用ssc.stop()方法来终止程序的执行。
对于Spark Streaming从EventHub接收数据的应用场景，可以用于实时处理和分析来自各种数据源的大规模数据流，例如传感器数据、日志数据、社交媒体数据等。通过使用Spark Streaming和EventHub，可以实现高吞吐量、低延迟的实时数据处理。
腾讯云提供了类似的云计算产品，例如腾讯云流计算（Tencent Cloud StreamCompute），可以用于实时数据处理和分析。您可以访问腾讯云的官方网站了解更多关于腾讯云流计算的信息：腾讯云流计算产品介绍，腾讯云流计算文档。
相关·内容
我工作的火花流基本上是从EventHub读取接近实时的数据，并转储到斑点位置，我将需要实现水印，以查看什么是延迟(数据滞后) b/w的EventHub和斑点位置。正如我们所知，EventHub只保存2天的数据，我需要确保没有数据丢失b/w EventHub和blob位置。有没有办法在Spark structured streaming中实现这一点？
浏览 13提问于2019-11-06得票数 0
在我的spark streaming应用程序中，我尝试从Azure EventHub中流式传输数据，并根据数据写入到hdfs blob中的几个目录中。基本上遵循的是链接
代码如下：
def writeStreamer(input: DataFrame, checkPointFolder: String, output: String): StreamingQuery = {
input
.writeStream
.format("com.databricks.spark.avro")
.partitionBy("year", "
浏览 0提问于2019-11-07得票数 0
当我运行spark streaming example org.apache.spark.examples.streaming.JavaDirectKafkaWordCount时，我捕捉到一个EOFException follow，我如何解决它
Exception in thread "main" org.apache.spark.SparkException: java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
java.io.EOFExcept
浏览 5提问于2015-06-19得票数 1
3回答
我有一个spark流作业，它每5秒从Kafka读取一次，对传入的数据进行一些转换，然后写入文件系统。
这真的不需要是一个流作业，实际上，我只想每天运行一次，将消息排出到文件系统中。不过，我不确定如何停止这项工作。
如果我向streamingContext.awaitTermination传递一个超时，它不会停止进程，它所做的只是在迭代流时导致进程产生错误(参见下面的错误)。
完成我正在尝试做的事情的最佳方法是什么
这是针对Python上的Spark 1.6的
编辑：
感谢@marios，解决方案是这样的：
ssc.start()
ssc.awaitTermination(10)
ssc.sto
浏览 2提问于2016-01-30得票数 18
回答已采纳
3回答
我在通过spark streaming从azure blobs读取数据时遇到问题
JavaDStream<String> lines = ssc.textFileStream("hdfs://ip:8020/directory");
类似上面的代码适用于HDFS，但无法从Azure blob读取文件
https://blobstorage.blob.core.windows.net/containerid/folder1/
上面是在azure UI中显示的路径，但这不起作用，我是不是遗漏了什么，以及我们如何访问它。
我知道Eventhub是流式传输数据的理想选择，但
浏览 4提问于2016-06-11得票数 11
回答已采纳
我使用的是spark 2.3.2，在对来自Kafka的2个或更多流媒体资源进行联合时遇到了一个问题。这些都是来自Kafka的流媒体源，我已经将它们转换并存储在Dataframes中。 理想情况下，我希望将此UNIONed数据帧的结果以parquet格式存储在HDFS中，甚至可能存储回Kafka中。最终目标是以尽可能低的延迟存储这些合并的事件。 val finalDF = flatDF1
.union(flatDF2)
.union(flatDF3)
val query = finalDF.writeStream
.format("parquet&
浏览 52提问于2019-07-02得票数 1
回答已采纳
1回答
采用火花流的实时异常检测系统。在每个流间隔中，如果数据点异常，AWS SNS会发送一封电子邮件给订阅帐户。但是AWS SNS java sdk喜欢在火花流中不工作。下面是错误消息
Error StreamingContext:错误启动上下文，将其标记为已停止的java.io.NotSerializableException: DStream检查点已启用，但具有其功能的DStreams不是可序列化的com.amazonaws.services.sns.AmazonSNSClient序列化堆栈：- object不可序列化( class : com.amazonaws.services.sns.A
浏览 1提问于2016-01-04得票数 0
回答已采纳
1回答
我必须用下面的用例设计一个火花流应用程序。我正在寻找最好的方法来解决这个问题。
我有将数据推入1000+的应用程序，不同的主题，每个主题都有不同的用途。火花流将接收来自每个主题的数据，处理后它将写回相应的另一个主题。
Ex.
Input Type 1 Topic --> Spark Streaming --> Output Type 1 Topic
Input Type 2 Topic --> Spark Streaming --> Output Type 2 Topic
Input Type 3 Topic --> Spark Streaming
浏览 2提问于2017-06-13得票数 2
我正在集群模式下在纱线上运行一个Spark流应用程序，并且我正在尝试实现一个优雅的关闭，以便当应用程序被关闭时，它将在停止之前完成当前微批处理的执行。
在一些教程之后，我将spark.streaming.stopGracefullyOnShutdown配置为true，并将以下代码添加到我的应用程序中：
sys.ShutdownHookThread {
log.info("Gracefully stopping Spark Streaming Application")
ssc.stop(true, true)
log.info("Applicati
浏览 5提问于2016-04-28得票数 9
我正在尝试将xml文件(小于100 kb)发送到Azure事件中心，然后在发送它们之后，读取Databricks中的事件。
现在，我已经使用Python以字节发送XML的内容(这个步骤是工作)。但是我想要完成的下一步是从事件的“主体”中读取该XML内容，并使用PYSPARK创建一个Spark。
要做到这一点，我有两个疑问：
XML1-我在选项中指定事件的“主体”的内容是XML的选项吗? 2-是否有其他方法可以直接将该内容转储到Spark ? 3-在将spark.readStream作为事件发送时缺少一些配置吗？
我试着像下面的例子一样：
Python事件生成器
# this is the p
浏览 0提问于2021-01-20得票数 1
回答已采纳
1回答
在Spark structutred streaming (2.3)中，是否允许连接来自同一输入流数据集的两个流？
例如，在下面的示例查询中，连接了两个流。我在Azure eventhub spark客户端中获得了IllegalStateException。
这会起作用吗？
eventhubs = spark.readStream ... .createOrReplaceTempView("Input")
spark.sql("SELECT temperature, time, device, category FROM Input").createOrRe
浏览 1提问于2018-07-29得票数 1
--我知道类似的问题已经得到了简短的回答，但由于缺乏最低限度的reputation...hence，我无法在此补充我个人的怀疑--我在这里问它--
我想使用Apache + Kafka来处理Twitter数据。我为此创造了一个模式。但是当我运行它时，我会得到以下错误。我搜索了很多关于这个错误的地方，但是我无法得到我想要的解决方案，或者它没有工作。上一次我用较小的内存空间运行Spark时，我认为内存不足，但我仍然会遇到同样的错误。这是我收到错误的代码：
from kafka import KafkaConsumer
from pyspark.streaming import StreamingC
浏览 2提问于2019-12-29得票数 2
回答已采纳
我有一条从卡夫卡读到GCP的管道。档案记录太少了。我想创建更大的文件。到目前为止，这是我如何配置梁(至少我认为是一个相关的参数)。我的问题是如何控制从束流管道产生的文件的大小？
windowDuration: 5
numShards: 0
batchIntervalMillis: 30000
checkpointDurationMillis: 30000
maxRecordsPerBatch: 60000000
以下是与流相关的火花配置参数。
spark.default.parallelism=600
spark.ui.retainedStages=10
sp
浏览 3提问于2021-05-26得票数 0
回答已采纳
我使用的是最新安装的spark (2.1.0)和python (3.5.3)。我在本地安装了kafka (2.10.0)。
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pykafka import KafkaClient
import json
import sys
import pprint
spsc = SparkContext(appName="SampleA
浏览 0提问于2017-03-26得票数 0
我正在创建一个使用火花SQL (数据帧)和火花流的演示。我不是火花专家，所以我需要一些帮助！
我从一个数据库加载了大约100万个对象到spark Dataframe，我执行SQL查询来匹配一些字段和来自spark streaming的实时数据。
例如,
SELECT *
FROM Person
WHERE Person.name='stream.name' AND Person.age='stream.age' AND ... etc
stream.xxx是我从spark streaming RDD中提取出来的一个java字符串。
现在，问题是，对于具有100万
浏览 1提问于2015-06-27得票数 2
1回答
我正在尝试将数据写入一个卡夫卡主题后，阅读了一个蜂巢表如下。
write_kafka_data.py:
read_df = spark.sql("select * from db.table where some_column in ('ASIA', 'Europe')")
final_df = read_df.select(F.to_json(F.struct(F.col("*"))).alias("value"))
final_df.write.format("kafka")\
浏览 0提问于2021-09-13得票数 4
回答已采纳
1回答
我构建了纱线集群(Docker)来运行java基SparkStreaming。
我不知道为什么mapPartitions在KafkaStreaming上重新运行相同的函数。
executor1在本地传输。
我找到了两次从远程传输数据的执行者2。
我需要配置一些东西让same不能重新运行相同的功能吗？
```javascript
JavaPairInputDStream<String，String> inputDStream = KafkaUtils.createStream(jssc，String.class，String.class，StringDecoder.class，St
浏览 1提问于2017-06-27得票数 0
我是Spark streaming的新手。我正在尝试使用本地csv文件进行结构化Spark流。我在处理的时候遇到了下面的异常。
Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start();;
FileSource[file:///home/Teju/Desktop/SparkInputFiles/*.csv]
这是我的代码。
val df = spark
.
浏览 0提问于2017-07-28得票数 0
我想使用星火流，并连接它与卡夫卡。然而，我仍然得到了NoSuchMethodError: NoSuchMethodError错误，现在我只是不知道下一步该做什么。
我的设置：
Ubuntu 16.04
Scala 2.11
Kafka 2.11-1.0.0 (我也尝试使用2.11-0.10.0.0)
火花2.2.1
Hadoop 2.9.0
我甚至不能运行示例脚本：
from pyspark.sql import SparkSession, Row, SQLContext
from pyspark import SparkContext, SparkConf
from pyspark.stre
浏览 1提问于2018-02-09得票数 3
我使用下面的自定义接收器来使用Scala中Rabbitmq中的数据。
import org.apache.spark.streaming.rabbitmq.RabbitMQUtils
import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket
import java.nio.charset.StandardCharsets
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
imp
浏览 11提问于2022-01-06得票数 0
1回答
我必须为我的流作业编写一个输出文件。
问:我的工作什么时候才能真正停止？我杀了服务器，但没有工作。我想从命令行停止我的工作(如果可能的话)
代码：
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.dstream
import org.apache.spark.streaming.Duration
import org.apache
浏览 2提问于2017-04-13得票数 0
回答已采纳
我正在构建一个卡夫卡吞食模块在EMR 5.11.1，火花2.2.1。我的意图是使用结构化流从一个Kafka主题消费，做一些处理，并存储到EMRFS/S3的拼花格式。
控制台接收器按预期工作，文件接收器不工作。
在spark-shell中
val event = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", <server list>)
.option("subscribe", <topic>)
.load()
val eventdf
浏览 0提问于2018-04-09得票数 13
我希望使用火花结构流从卡夫卡读取数据，并处理它，并写入蜂窝表。
val spark = SparkSession
.builder
.appName("Kafka Test")
.config("spark.sql.streaming.metricsEnabled", true)
.config("spark.streaming.backpressure.enabled", "true")
.enableHiveSupport()
.getOrCreate()
val events = sp
浏览 5提问于2018-02-06得票数 1
我正在尝试使用spark streaming来读取kafka主题中的数据。我可以将消息生成kafka主题，但当使用spark streaming读取主题中的数据时，我收到错误消息，如下所示：
ERROR ReceiverTracker: Deregistered receiver for stream 0: Error starting receiver 0 - java.lang.ClassCastException: java.util.HashMap cannot be cast to java.lang.String
代码如下：
from pyspark import SparkCon
浏览 0提问于2019-05-27得票数 1
我想在特定时间后终止spark映射。我使用sqlContext.streams.awaitAnyTermination(long timeoutMs)来实现这一点。但是映射在给定的超时后不会停止。
我尝试从azure事件中心读取，并提供了2分钟(120000毫秒)作为awaitAnyTermination方法的超时。但映射不会在azure databricks群集上停止。
下面是我的代码。我在从azure eventhub上读东西，在控制台上写东西，在awaitAnyTermination上写12万毫秒。
import org.apache.spark.eventhubs._
// Even
浏览 2提问于2019-09-10得票数 0
1回答
我试图使用Twitter来发送tweet，但我不知道如何自动停止流，除非使用键盘中断( control+c)。
import tweepy
import time
from tweepy import Stream
from tweepy import OAuthHandler
from tweepy.streaming import StreamListener
class listener(StreamListener):
def on_data(self, data):
#print data
tweet = data.split(',&#
浏览 4提问于2014-03-08得票数 0
回答已采纳
1回答
我们将数据文件放在HDFS路径中，该路径由spark streaming应用程序监控。以及spark streaming应用程序向kafka topic发送数据。我们正在中间停止流应用程序？，然后再次启动，以便它应该从停止的位置开始。但是它正在重新处理整个输入数据文件。所以我猜检查点没有被正确使用。我们使用的是spark 1.4.1版本，如何才能使流媒体应用程序从失败/停止的位置启动？提前谢谢。
浏览 1提问于2015-12-23得票数 0
我用的是这样的东西：
df.withColumn("idx", monotonically_increasing_id())
但我得到了一个例外，因为它不被支持：
Exception in thread "main" org.apache.spark.sql.AnalysisException: Expression(s): monotonically_increasing_id() is not supported with streaming DataFrames/Datasets;;
at org.apache.spark.sql.catalyst.an
浏览 3提问于2021-01-05得票数 1
我构建了一个火花流应用程序来继续接收来自Kafka的消息，然后将它们写到一个表HBase中。
这个应用程序在前25分钟运行得很好。当我在Kafka-console-producer中输入KV对(如1;name1、2;name2 )时，可以将它们保存在Hbase表中：
ROW COLUMN+CELL
1 column=cf1:column-Name, timestamp=1471905340560, value=name1
2 column=cf1:column-Name, timestamp=1471905348165, value=name2
浏览 4提问于2016-08-29得票数 2
回答已采纳
2回答
我在ApacheSpark2.2中使用了最新的结构化流，并得到了以下例外：
org.apache.spark.sql.AnalysisException:当流数据框架/数据集上没有流聚合时，不支持完整的输出模式；
为什么完全输出模式需要流聚合？如果Spark允许流查询中没有聚合的完整输出模式，会发生什么情况？
scala> spark.version
res0: String = 2.2.0
import org.apache.spark.sql.execution.streaming.MemoryStream
import org.apache.spark.sql.SQLCo
浏览 1提问于2017-08-18得票数 19
回答已采纳
1回答
卡夫卡的电火花印刷信息
、、、
我建立了一个由生产者和消费者组成的kafka系统，将json文件的行作为消息流。
使用pyspark，我需要分析不同流窗口的数据。要做到这一点，我需要看一看数据，因为它们是由火花放电流.我该怎么做呢？
为了运行代码，我使用了容器。下面是我的python代码：
# Add dependencies and load modules
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--conf spark.ui.port=4040 --packages org.apache.spark:spark-streaming-kafka
浏览 0提问于2018-11-18得票数 1
回答已采纳
我想从Azure甚至获取1天的数据，并应用一些逻辑并将其复制到cosmos DB中。我可以从甚至暴徒中获取数据，但是数据正在流。我只需要在一个时间窗口中获取数据(比如只需要一天/或5小时)。
下面是我试图从Azure EventHub获取数据的代码。
import org.apache.spark.eventhubs.{ ConnectionStringBuilder, EventHubsConf, EventPosition }
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
浏览 1提问于2019-09-23得票数 0
我使用Spark2.2.0，并在windows上使用Spark结构化流时出现以下错误：
当没有streaming aggregations的streaming DataFrames/DataSets上有watermark时，不支持完整的输出模式。
浏览 0提问于2017-11-13得票数 2
回答已采纳
代码： df_streaming = spark \
.readStream \
.format("kafka") \
... \
.load() \
.xxx()
df_streaming = df_streaming \
.groupBy(["name", "height"]) \
.apply(cal_feature)
stream_writer = df_streaming \
.writeStream \
.format("console")
浏览 8提问于2021-06-16得票数 0
我正在尝试从Spark Structured Streaming中的S3存储桶中读取数据。下面的代码用于获取现有数据。但是，当新数据添加到存储桶中时，Spark不会选择这一点。
val lines = spark.readStream.schema(schemaImp).format("com.databricks.spark.avro").load("s3n://bucket/*")
val query = lines.writeStream.outputMode("append").format("memory").query
浏览 0提问于2016-12-10得票数 3
1回答
带有火花流的Kafka抛出了一个错误：
from pyspark.streaming.kafka import KafkaUtils ImportError: No module named kafka
我已经建立了一个卡夫卡经纪人和一个工作火花环境与一个主人和一个工人。
import os
os.environ['PYSPARK_PYTHON'] = '/usr/bin/python2.7'
import findspark
findspark.init('/usr/spark/spark-3.0.0-preview2-bin-hadoop2.7&#
浏览 2提问于2020-05-19得票数 4
回答已采纳
我正在尝试使用Spark Structured Streaming将聚合数据写入Kafka。下面是我的代码：
dataset
.writeStream()
.queryName(queryName)
.outputMode(OutputMode.Append())
.format("kafka")
.option("kafka.bootstrap.servers", kafkaBootstrapServers)
.option("topic", "topic")
.trigg
浏览 1提问于2018-04-10得票数 1
我正在尝试通过帮助twitter-streaming.But来保存来自twitter的保存推文，但我有一个问题:我的程序在一段时间后停止工作(取决于接近4-5秒的1毫秒的批处理间隔)。所以，你能帮我解决这个问题吗)。请告诉我出了什么问题？
当批处理间隔接近100毫秒时，我看到一些记录，如
19/08/06 23:45:26 INFO BlockRDD: Removing RDD 103 from persistence list
19/08/06 23:45:26 INFO BlockManager: Removing RDD 103
19/08/06 23:45:26 INFO Twitte
浏览 7提问于2019-08-07得票数 0
回答已采纳
我想了解"spark.streaming.blockInterval"在星火流DirectAPI中扮演了什么角色，据我理解，"spark.streaming.blockInterval"用于计算分区，即#partitions = (receivers x* batchInterval) /blockInterval，但在DirectAPI中，火花流分区等于no。卡夫卡隔断。
"spark.streaming.blockInterval"是如何在DirectAPI中使用的？
浏览 0提问于2016-12-17得票数 0
回答已采纳
2回答
火花流写问题
、、
我试图从文本文件中的JSON记录中创建一个动态模式，因为每个记录都有不同的模式。下面是我的代码。
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.functions.{lit, schema_of_json, from_json, col}
object streamingexample {
def main(args: Array
浏览 5提问于2020-03-10得票数 0
4回答
我有一个Spark Streaming作业，它一直在持续运行。如何优雅地停止作业？我已经阅读了在作业监视中附加关闭钩子并向作业发送SIGTERM的常用建议。
sys.ShutdownHookThread {
logger.info("Gracefully stopping Application...")
ssc.stop(stopSparkContext = true, stopGracefully = true)
logger.info("Application stopped gracefully")
}
这似乎是有效的，但看起来不是停止工作的
浏览 0提问于2015-09-15得票数 30
当我提交一个jar文件时没有错误。
但是，当我使用HTTP协议发送数据时，不会打印数据。
(当我使用“kafka-console- well . is”检查数据时，数据打印得很好)
图片，提交了一个jar文件:数据没有打印
jar文件中的代码和依赖项如下所示。
图片，卡夫卡-控制台-消费. is :数据打印
指挥：
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --group test-consumer --topic test01 --from-beginning
JAVA文件
2-1，属地
&
浏览 5提问于2021-05-04得票数 0
2回答
我正在使用Spark2.0并尝试使用wholeTextFiles API来流文件。我的Spark程序正在成功地读取文件--文件夹中的第一批文件，但是我不能流到后面的一批文件。
请让我知道如何在使用WholeTextFiles API时流文件。
这是我的代码：
SparkConf sparkConf = new SparkConf().setAppName("My app")
.setMaster("local")
.set("spark.driv
浏览 4提问于2017-04-05得票数 0
1回答
目标：不断地将嗅探的网络包提供给卡夫卡生产商，将其与星火流连接起来，以便能够处理包数据，然后使用Tensorflow或Keras中的预处理数据。
我正在处理来自卡夫卡的火花流(PySpark)中的连续数据，现在我想将处理过的数据发送到Tensorflow。如何用Python在Tensorflow中使用这些转换后的DStreams？谢谢。
目前没有任何处理应用于星火流，但将在以后添加。以下是py代码：
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from py
浏览 0提问于2018-12-18得票数 3
2回答
我是新的火花和卡夫卡和我有一个稍微不同的使用模式火花流与卡夫卡。我在用
spark-core_2.10 - 2.1.1
spark-streaming_2.10 - 2.1.1
spark-streaming-kafka-0-10_2.10 - 2.0.0
kafka_2.10 - 0.10.1.1
连续事件数据正在被流到一个kafka主题中，我需要从多个火花流应用程序中处理这个主题。但是当我运行星火流应用程序时，只有其中一个接收数据。
Map<String, Object> kafkaParams = new HashMap<String, Object>(
浏览 3提问于2017-08-29得票数 4
回答已采纳
我想使用结构化流和PySpark一起运行Spark应用程序。
我使用星火2.2和卡夫卡0.10版本。
如果出现以下错误，我将失败：
java.lang.IncompatibleClassChangeError:实现类
spark-submit命令，如下所示：
/bin/spark-submit \
--packages org.apache.spark:spark-streaming-kafka-0-10_2.11:2.2.0 \
--master local[*] \
/home/umar/structured_streaming.py localhost:2181 fo
浏览 5提问于2017-08-22得票数 3
1回答
火花流
、
我编写了这段代码，用于吡火花上的虹膜分类，但是我得到了一个错误"'RDD‘对象没有属性'_jdf’“。我已经将RDD更改为dataframe，但它告诉我们，"RDD是不可移植的“。请帮我解决！非常感谢。
这里是我的代码：
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.ml import PipelineModel, Pipeline
from pyspark.sql import Row, DataFrame
from py
浏览 0提问于2018-08-17得票数 2
1回答
将Spark流数据发送回客户端
、、、、
我是Apache Spark Streaming的新手。我正在开发一个spark流应用程序，以找到最短路径，并再次将该路径发送回客户端。我已经编写了获取数据并使用函数处理数据的代码，但我有一个问题，如何将结果再次发送回客户端以下是我的代码： import networkx as nx
from pyspark import SparkConf,SparkContext
from pyspark.streaming import StreamingContext
TCP_IP = "127.0.0.1"
TCP_PORT = 5000
# Creating a Spark C
浏览 11提问于2020-04-27得票数 0
回答已采纳
1回答
无数据时停止提交作业
、、、
我使用spark streaming通过自定义接收器从mq中的队列中获取数据。Javastreaming上下文持续时间为10秒。并且为来自队列的输入定义了一个任务。
在spark UI中的事件时间线中，我看到每隔10秒就会提交一个作业，即使没有来自接收方的数据。
这是正常行为吗?或者在没有数据时如何停止提交作业？
JavaDStream<String> customReceiverStream = ssc.receiverStream(newJavaCustomReceiver(host, port));
JavaDStream<String> words =line
浏览 2提问于2018-03-15得票数 0
我试着用spark连接到kafka的话题。它不会读取数据流中的任何数据，也不会产生任何错误。下面是我的jupyter代码：
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'
from pretty import pprint
from pyspark import SparkContext
from pyspark.streaming import Stream
浏览 0提问于2019-12-19得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
