在Spark Streaming中处理太迟的数据
是指数据到达Spark Streaming系统时已经超过了指定的处理时间窗口。这种情况可能会导致数据处理的延迟，影响实时性和准确性。
为了解决处理太迟的数据的问题，可以采取以下几种方法：
- 调整批处理时间窗口：通过增加批处理时间窗口的大小，可以容纳更多的数据并提高处理的实时性。但是需要权衡处理延迟和系统资源的消耗。
- 增加集群资源：通过增加集群的计算和存储资源，可以提高数据处理的速度和容量，减少处理延迟。可以根据实际需求调整集群规模和配置。
- 优化数据传输和存储：使用高效的数据传输和存储方式，如使用压缩算法减小数据传输量、使用分布式文件系统提高数据读写速度等，可以减少数据处理的延迟。
- 使用窗口操作和状态管理：Spark Streaming提供了窗口操作和状态管理的功能，可以对数据进行分组和聚合，并维护状态信息。通过合理设置窗口大小和滑动间隔，可以更好地处理延迟数据。
- 实时监控和调优：通过监控系统的实时指标和日志信息，及时发现和解决处理延迟的问题。可以使用Spark的监控工具和日志分析工具进行系统调优和故障排查。
在处理太迟的数据时，腾讯云提供了一系列的云计算产品和服务，可以帮助用户实现高效的数据处理和分析。例如，腾讯云的云服务器、云数据库、云存储等基础服务可以提供高性能的计算和存储资源；腾讯云的云原生服务、人工智能服务、物联网服务等可以提供丰富的功能和工具，帮助用户构建和管理复杂的云计算应用。具体产品和服务的介绍和链接地址可以参考腾讯云官方网站。
相关·内容
我正在尝试使用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
在带有检查点的Spark结构化流应用程序中，我得到了“偏移量从X更改为0，一些数据可能被漏掉”的错误，但它似乎并没有真正造成任何问题。我想弄清楚这个错误到底意味着什么。
我的设置如下。
我让Kafka (0.10.1.0)在一个码头容器中运行，在/tmp/kafka日志上挂载了一个命名的卷，以便在重新启动之间保持日志。
我在另一个码头容器中有一个Spark结构化流(2.1.1)应用程序。溪流消耗卡夫卡的数据。它们还在重新挂载到指定卷中的位置使用检查点，以确保在重新启动之间保留元数据。
我使用一个实现接口的自定义接收器，这意味着我必须实现自己的已处理版本日志，以便当一切重新启动时，我
浏览 2提问于2017-05-09得票数 3
我在yarn集群上使用spark-streaming (spark version 2.2)，并尝试为我的应用程序启用动态内核分配。 executor的数量可以根据需要增加，但是一旦分配了executor，即使流量减少了，它们也不会缩小，也就是说，一旦分配了executor就不会被释放。我还在yarn上启用了external shuffle服务，如下所述：https://spark.apache.org/docs/latest/running-on-yarn.html#configuring-the-external-shuffle-service 我在spark-submit命令中设置的配
浏览 19提问于2018-12-25得票数 1
回答已采纳
我的理解是Spark structured Streaming是建立在Spark SQL之上的，而不是Spark streaming。因此，下面的问题是，应用于spark流的属性是否也适用于spark结构化流，例如：
spark.streaming.backpressure.initialRate spark.streaming.backpressure.enabled spark.streaming.receiver.maxRate
浏览 1提问于2018-09-03得票数 1
我在运行无法连接到Kinesis数据源的胶水流式作业时遇到以下错误：
错误：
WARNING:root:StreamingQueryException caught. Retry number 10 ERROR:root:Exceeded maximuim number of retries in streaming interval,
exception thrown Parse yarn logs get error message:
StreamingQueryException: 'Error while Describe Streams\n=== Streaming Q
浏览 0提问于2020-09-08得票数 0
我正在从kafka流式传输数据，并试图将每批事件的数量限制在10个事件。在处理10-15批次后，批次大小突然出现峰值。下面是我的设置：
spark.streaming.kafka.maxRatePerPartition=1
spark.streaming.backpressure.enabled=true
spark.streaming.backpressure.pid.minRate=1
spark.streaming.receiver.maxRate=2
浏览 4提问于2017-11-17得票数 0
我已经开发了一个对文件流进行检查点的Spark流应用程序。我需要在任何驱动器上停止我的流媒体应用程序exception..my代码如下：
val fileStream=..
fileStream.checkpoint(Duration(batchIntervalSeconds * 1000 * 5))
//initiate the chekpointing
fileStream.foreachRDD(r=> {
try {
r.count()
} catch {
case ex: Exception => {
ssc.stop(true,
浏览 8提问于2016-09-28得票数 3
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
我正在尝试使用ZKStringSerializer类，这是我通过
import kafka.utils.ZKStringSerializer
根据整个互联网，甚至在我被计算机重启之前我自己的代码，这应该可以让我的代码工作。然而，我现在得到了一个令人难以置信困惑编译错误，
object ZKStringSerializer in package utils cannot be accessed in package kafka.utils
这很让人困惑，因为这个文件不应该在任何包中，而且我也没有在任何地方指定包。这是我的代码：
import org.apache.spark.sql.SparkS
浏览 7提问于2018-07-03得票数 1
我正在尝试从Kafka读取JSON消息，并使用spark structured将它们存储在HDFS中。
我遵循了示例，当我的代码如下所示时：
df = spark \
.read \
.format("kafka") \
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \
.option("subscribe", "topic1") \
.load()
df.selectExpr("CAST(key AS STR
浏览 43提问于2018-07-25得票数 1
我将一行从createStream改为createDirectStream，因为新库不支持createStream
我已经从这里检查过了，
scala> val lines = KafkaUtils.createDirectStream(ssc, "localhost:2181", "sparkgroup", topicpMap).map(_._2)
<console>:44: error: overloaded method value createDirectStream with alternatives:
[K, V
浏览 6提问于2022-05-15得票数 1
我有一个spark-streaming服务，在那里我正在处理和检测基于某个离线生成的模型的异常。我从日志文件向此服务提供数据，日志文件使用以下命令进行流式传输
tail -f <logfile>| nc -lk 9999
在这里，spark流服务从端口9999获取数据。但是，我观察到最后几行正在被丢弃，即spark streaming没有接收到这些日志行，或者它们没有被处理。
但是，我也注意到，如果我只是将日志文件作为标准输入，而不是对其进行尾随，则不会删除任何行：
nc -q 10 -lk 9999 < logfile
有人能解释为什么会发生这种行为吗？对于将日志数据流式传
浏览 0提问于2016-02-09得票数 1
我正在尝试使用上面提到的技术设计一个IoT平台。如果有人能评论这个架构，我会很高兴，如果它是好的和可伸缩的！
我通过mqtt获取IoT传感器数据，我将通过spark streaming接收这些数据(有一个用于spark streaming的mqtt连接器，它可以做到这一点)。我只需要订阅主题，并且有一个第三方服务器将IoT数据发布到主题。
然后我解析数据，并将其插入到亚马逊网络服务DynamoDB中。是的，整个安装程序将在AWS上运行。
根据IoT用例，我将来可能需要处理/转换数据，所以我认为spark可能会很有用。我也听说spark流媒体的速度很快。
这是一个简单的概述，我不确定它是否是一个
浏览 1提问于2017-06-23得票数 0
3回答
我有一个spark流作业，它每5秒从Kafka读取一次，对传入的数据进行一些转换，然后写入文件系统。
这真的不需要是一个流作业，实际上，我只想每天运行一次，将消息排出到文件系统中。不过，我不确定如何停止这项工作。
如果我向streamingContext.awaitTermination传递一个超时，它不会停止进程，它所做的只是在迭代流时导致进程产生错误(参见下面的错误)。
完成我正在尝试做的事情的最佳方法是什么
这是针对Python上的Spark 1.6的
编辑：
感谢@marios，解决方案是这样的：
ssc.start()
ssc.awaitTermination(10)
ssc.sto
浏览 2提问于2016-01-30得票数 18
回答已采纳
为了避免延迟和加快进程，我在spark流中构建了线程池。主程序列表如下：
stream.foreachRDD(rdd=> {
rdd.foreachPartition { rddPartition => {
val client: Client = ESClient.getInstance.getClient
var num = Random.nextInt()
val threadPool: ExecutorService = Executors.newFixedThreadPool(5)
val confs = new Configurati
浏览 0提问于2017-11-12得票数 1
1回答
在spark streaming中，流数据将由在workers上运行的接收器接收。数据将被定期推送到数据块中，接收器将receivedBlockInfo发送给驱动程序。我想知道spark streaming是否会将数据块分发到集群？(换句话说，它是否会使用分布式存储策略)。如果不跨集群分发数据，如何保证工作负载均衡？(假设我们有一个10个节点的集群，但只有少数几个接收器)
浏览 0提问于2015-06-05得票数 1
我正在分析日志，我有这样的架构：
kafka -> spark streaming ->弹性搜索
我的主要目标是在流媒体中创建机器学习模型。我认为我可以做两件事：
1) Kafka -> spark Streaming (ML) ->弹性搜索
2) Kafka -> spark Streaming-> elasticsearch -> spark streaming(ML)
-I认为第二种架构是最好的，因为spark streaming将直接使用索引数据。你认为如何？-If我们在spark streaming中创建一个模型(在弹性搜索之后)我们必须在这
浏览 1提问于2017-05-10得票数 4
我有水槽流，我想把它存储在HDFS通过火花。下面是我正在运行的火花代码
object FlumePull {
def main(args: Array[String]) {
if (args.length < 2) {
System.err.println(
"Usage: FlumePollingEventCount <host> <port>")
System.exit(1)
}
val batchInterval = Milliseconds(60000)
val
浏览 3提问于2016-04-01得票数 2
回答已采纳
我在Yarn上使用Spark Streaming，我面临着以下问题。
问题1：
我在yarn上使用spark streaming (1.6.1)，我总是看到活动作业计数为1，这意味着一次只有1个作业在运行。我已经使用了"--conf spark. streaming. concurrentJobs=3“参数，但不幸的是我总是只能看到1个活动作业。
问题2：
我有50个Kafka分区和spark streaming创建了50个RDD分区，但我可以看到95%的任务分配给了只有1个executor，其余的executor大多总是没有活动任务。
我的Spark Submit
浏览 0提问于2016-12-08得票数 0
1回答
我们在Hortonworks上使用Kafka和Spark streaming在本地集群设置中运行实时流应用程序。
我们有一个需求，我们需要从Spark Streaming或Kafka推送一些事件触发的数据，以保存在亚马逊网络服务的S3文件系统中。
任何关于这方面的建议都将不胜感激。
浏览 0提问于2017-02-13得票数 0
运行Spark1.6.2(纱线模式)
首先，我有一些从这篇文章到的代码，所以这可能是问题所在，但希望不是。
基本上，我有第一份工作。
import org.apache.spark.SparkContext
import org.apache.spark.streaming.{StreamingContext, Seconds}
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
def getStream(ssc: Stre
浏览 1提问于2017-02-01得票数 0
回答已采纳
我们讨论了以下问题：
但是Spark Structured Streaming是在Spark2.2上添加的，它给流媒体带来了很多变化，而且它是非常出色的。
我们可以说Spark Strutured Streaming是流处理，还是仍然是批处理？
现在Apache Flink和Apache Spark Structured Streaming的最大区别是什么？
浏览 3提问于2017-09-01得票数 13
我正在做一个数据分析项目，在这个项目中，我从CSV文件中读取数据，在Kafka主题上遍历该文件，并使用Spark Streaming来使用该Kafka主题数据。我在一个单一的项目中使用的所有组件。
现在，在使用Spark Streaming消费数据后，我必须对其进行一些计算，我必须将数据保存到elastic search中，并且我必须将该数据发送到另一个主题。所以我从Spark Streaming开始做这些事情(将数据保存到elastic中，并将数据发送到topic)。
下面是我的代码
@Component
public class RawEventSparkConsumer implemen
浏览 0提问于2019-06-04得票数 0
1回答
如何在应用转换之前将每个火花输入流中的数据集组合为一个。我用的是火花-2.0.0
val ssc = new StreamingContext(sc, Seconds(2))
val sqlContext = new SQLContext(sc)
val lines = ssc.textFileStream("input")
lines.foreachRDD { rdd =>
val count = rdd.count()
if (count > 0) {
val dataSet = sqlContext.read.json(rdd)
浏览 2提问于2016-08-08得票数 0
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
回答已采纳
我们有一个在Spark2.3.3上运行的星火流应用程序
基本上，它打开了一个卡夫卡流：
kafka_stream = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "mykafka:9092") \
.option("subscribe", "mytopic") \
.load()
kafka主题有两个分区。之后，在列上有一些基本的过滤操作、一些Python和basic ()，例如：
浏览 1提问于2019-04-02得票数 10
回答已采纳
1回答
我使用结构化流从Kafka读取数据，并创建各种聚合度量。我已经使用metrics.properties启用了石墨接收器。我已经看到应用程序在较早的星火版本有流相关的指标。我没有看到与结构化流相关的流标准。我做错了什么？
例如，无法找到未处理的批处理或正在运行的批处理或最后完成的批处理总延迟。
我通过设置以下方法启用了流度量：
SparkSession.builder().config("spark.sql.streaming.metricsEnabled",true)
即使如此，我也只能得到三个指标：
driver.spark.streaming.inputrate
d
浏览 4提问于2017-12-19得票数 12
默认情况下，结构化流作业使用HDFSStateStoreProvide。使用HDFS存储的问题是它是不可伸缩的。当作业在交通繁忙时间从kafka获得更多数据时，由于以下错误而失败：
18/12/06 15:54:35 ERROR scheduler.TaskSetManager: Task 191 in stage 231.0 failed 4 times; aborting job
18/12/06 15:54:35 ERROR streaming.StreamExecution: Query eventQuery [id = 42051afe-b1bc-438d-8143-2d7e5def
浏览 1提问于2018-12-07得票数 1
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
1回答
火花流待批处理
、、
我正在运行一个Spark Streaming应用程序，它从Kafka (使用Direct Stream方法)读取数据，并将结果发布回Kafka。该应用程序的输入速率以及应用程序的吞吐量在大约一两个小时内保持稳定。之后，我开始看到在Active Batches队列中保留很长时间(对于30mins+)的批处理。Spark driver日志指示以下两种类型的错误，并且这些错误发生的时间与批处理的开始时间非常吻合：
第一错误类型
ERROR LiveListenerBus: Dropping SparkListenerEvent because no remaining room in event q
浏览 3提问于2017-08-04得票数 0
在spark streaming中，接收到的数据在集群中工作节点的多个Spark executors之间复制(默认复制系数为2)()。但是，如何才能获得特定RDD的复制位置？
浏览 2提问于2015-06-17得票数 0
我使用的是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
回答已采纳
在spark streaming中，我们接收的DStreams是一批RDDs。那么窗口如何进一步提供帮助呢？
据我所知，它还批处理RDDs。
如果我错了，请纠正我( Spark Streaming的新手)。
浏览 4提问于2015-10-08得票数 7
回答已采纳
我使用直接的Kafka API以1分钟的批处理间隔运行我的spark流媒体应用程序，并在我的应用程序代码中使用Pandas和Pyspark。
下面是我的集群配置:3个数据节点，每个数据节点具有8核、12 my的容量。
我已经提供了如下参数的spark-submit作业，
--master yarn
--deploy-mode cluster
--executor-memory 2G
--total-executor-cores 4
--num-executors 11
但是我的SPARK UI显示我的活动批处理处于队列状态，
config("spark.streaming.backp
浏览 14提问于2017-08-28得票数 1
最近，我们将我们的项目从Spark 2.2.0 cloudera2迁移到Spark 2.3.0 cloudera2，并注意到一些以前可以工作的客户Sink，但现在失败了。为了简单起见，我重写了一个小案例，以便助手可以复制和粘贴代码来测试它。
package question
import java.io.PrintWriter
import java.net.Socket
import org.apache.spark.sql._
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.sou
浏览 0提问于2018-04-20得票数 0
1回答
我正在为一个用python编程的项目使用kafka和spark streaming。我想将kafka producer的数据发送到我的流媒体程序中。当我使用指定的依赖关系执行以下命令时，它可以顺利地工作：
./spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.1.0 ./kafkastreaming.py
是否有任何方法可以指定依赖项并直接运行流代码(例如，不使用spark-submit或使用spark-submit但不指定依赖项)。
我尝试在spark的conf目录中指定spark-defaul
浏览 2提问于2017-06-26得票数 0
1回答
我使用SparkStreaming1.5.2，并使用Direct方法摄取Kafka 0.8.2.2中的数据。
我启用了检查点，这样我的驱动程序就可以重新启动，并在不丢失未经处理的数据的情况下拾取中断的位置。
检查点是写入S3的，因为我在Amazon上，而不是在Hadoop集群之上运行。
批处理间隔为1秒，因为我想要一个低延迟。
问题是，将一个检查点写入S3需要1到20秒。它们正在内存中备份，最终应用程序失败了。
2016-04-28 18:26:55,483 INFO [org.apache.spark.streaming.CheckpointWriter] [pool-16-thread-
浏览 0提问于2016-05-02得票数 4
显然在Spark streaming中没有对Cassandra接收器的内置支持。我在网上找到了这个例子，它基于ForEachWriter为Spark structured streaming实现了一个自定义的Cassandra接收器： https://dzone.com/articles/cassandra-sink-for-spark-structured-streaming 我知道我们需要创建一个ForeachWriter实现，它负责打开到接收器(Cassandra)的连接，写入数据并关闭连接。所以CassandraSinkForeach和CassandraDriver类是有意义的。 但
浏览 31提问于2019-02-24得票数 0
使用下面的scala代码，我使用Spark结构化流分别处理来自Apache的和发送到Apache的数据流。
我可以使用kafka源成功地读取数据流，但是，在试图将流写入Kafka接收器时，我得到了以下错误：
ERROR MicroBatchExecution:91 - Query [id = 234750ca-d416-4182-b3cc-4e2c1f922724, runId = 4c4b0931-9876-456f-8d56-752623803332] terminated with error java.lang.IllegalArgumentException: Expected e.
浏览 0提问于2018-10-29得票数 4
回答已采纳
1回答
我在Spark Streaming 2.2.1工作中有16个接收器。过了一段时间，一些接收器处理的记录越来越少，最终每秒只能处理一条记录。可以在屏幕截图上观察到这些行为：
虽然我知道根本原因可能很难找到并且不明显，但有没有方法可以进一步调试这个问题？目前我不知道从哪里开始挖掘。会不会和背压有关？
Spark流属性：
spark.app.id application_1599135282140_1222
spark.cores.max 64
spark.driver.cores 4
spark.driver.extraJavaOptions -XX:+PrintFlagsFin
浏览 0提问于2020-09-24得票数 2
我用的是这样的东西：
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
据我所知，Spark可以使用Spark Streaming分析流。Kafka可以从多个来源接收数据。我不明白的是，如果我有一个Kafka集群从多个来源接收数据，数据会被发送到一个运行Spark Streaming的数据库吗？或者Spark Streaming是否在应用服务器上运行？
浏览 5提问于2017-01-25得票数 0
回答已采纳
我正在将一个Java项目从Cloudera 5.10升级到Cloudera 6.2。我们有Spark Streaming从Kafka读取数据来处理它并将结果写到其他地方。在升级过程中，Spark从v1.6升级到v2.1，Kafka从v0.8升级到v2.1。 为了执行流处理，我们使用KafkaUtils.createStream(...)连接到Kafka，但KafkaUtils在Kafka 2.11中不再可用。然而，我似乎找不到任何Spark Streaming + Kafka在Java中不使用此方法的示例或文档。 我是不是漏掉了什么？在这些版本中，连接两个世界的最佳方式是什么？
浏览 11提问于2021-09-27得票数 0
我试着测试火花流。我有独立的cloudera快速启动vm。使用以下命令启动星火壳：
spark-shell --master yarn-client --conf spark.ui.port=23123
在火花壳中，我执行了以下语句：
sc.stop()
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
val conf = new SparkConf().setAppName("Spark Streaming")
val ssc = new Streamin
浏览 2提问于2017-10-26得票数 1
回答已采纳
我基本上是想使用Kafka中的数据，并将其写入HDFS。但是发生的情况是，它没有在hdfs中写入任何文件。它会创建空文件。
也请指导我，如果我想在hdfs中写avro格式，我可以如何修改代码。
为了简单起见，我写入本地C驱动器。
import org.apache.spark.SparkConf
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds
import o
浏览 4提问于2017-10-08得票数 0
我有我的spark应用程序，从Kafka读取数据并摄取到Kudu。它已经成功运行了近25个小时，并将数据摄入到了Kudu中。在那之后，我看到从kafka日志中选出了新的kafka分区的领导者。我的应用程序进入完成状态，并出现以下错误
org.apache.spark.SparkException: ArrayBuffer(kafka.common.NotLeaderForPartitionException, org.apache.spark.SparkException: Couldn't find leader offsets for Set([test,0]))
at org.a
浏览 20提问于2018-02-13得票数 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
1回答
我正在尝试使用eclipse IDE中的java代码通过kafka producer生成一些随机数据。我在kafka consumer中收到了相同的数据，它也是在同一个IDE中使用java代码创建的。我的工作依赖于流数据。所以，我需要spark streaming来接收kafka生成的随机数据。对于spark streaming，我在jupyter-notebook中使用python代码。要集成kafka和spark，必须在spark jar中添加“spark-streaming kafka-0-10_2.12-3.0.0.jar”文件。我还尝试在pyspark中添加jar文件。这是我的spa
浏览 0提问于2020-11-09得票数 1
我试图使用在DStream上执行ETL，但得到了以下错误。你能帮我修一下吗？谢谢。
KafkaCardCount.scala:56:28: value reduceByKey is not a member of org.apache.spark.streaming.dstream.DStream[Any]
[error] val wordCounts = etl.reduceByKey(_ + _)
[error] ^
[error] one error found
[error] (compile:compileIncrem
浏览 3提问于2018-01-14得票数 2
回答已采纳
1回答
我试图实现一个从Kafka读取流数据的Spark流应用程序。流数据是( key，value)对，形式为"String，int"，我想要计算每个键的平均值。数据的形式如下：
x,20
y,10
z,3
...
我希望以有状态的方式度量每个键的平均值。因此，我打算保存值的和以及它对应的键在映射函数中出现在State中的次数。
def mappingFunc(key: String, value: Option[Double], state: State[Double], count: State[Int]): (String, Double) = {
val su
浏览 0提问于2018-10-05得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
