如何在Spark Streaming中运行(流式) reduceByKey

文章/答案/技术大牛

发布

1回答

apache-spark、pyspark、spark-streaming

我使用Python API for Spark streaming中的textFileStream()方法在创建XML文件时读取它们，将它们映射到一个xml值中，从ElementTree中提取“有趣的”项并将它们放入一个字典( ElementTree：flatMap )中，然后我执行一个reduceByKey()来聚合每个键的计数。在减少时，我的字典中只剩下每个网络(密钥)的总数据包数。我的问题是我在流媒体上遇到了麻烦。它不是保存一个运行的总数，而是

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

1回答

使用Flume +火花流的示例单词计数应用程序

scala、apache-spark、spark-streaming、flume

下面是我使用Scala在spark.streaming中获取Flume事件和进程的代码。当尝试使用reduceBykey函数时，我得到以下编译错误： value reduceByKey is not a member of org.apache.spark.streaming.dstream.DStream我不认为这是一个依赖问题，我有其他简单的应用程序在使用reduceBykey的同一个Eclipse中工作。package com

浏览 0提问于2015-09-18得票数 1

回答已采纳

1回答

如何停止运行星火流应用程序？

apache-spark、spark-streaming

我的星火流工作一直在运行。我想以优雅的方式停下来。但是，如何在运行中的应用程序上更新此参数？

浏览 2提问于2016-10-12得票数 11

1回答

Spark Structured中流数据帧与批处理数据帧的结合

apache-spark、spark-structured-streaming

如何在Spark Structured streaming中将流式数据帧和批量数据帧组合在一起？

浏览 0提问于2017-09-21得票数 1

1回答

如何将这个简单的Spark Streaming代码转换成多线程代码？

scala、apache-spark、apache-kafka、spark-streaming

附加的代码只是使用Kafka和Spark Streaming的字数统计实现。如何在流式传输时让每个分区有单独的使用者执行？请帮帮我！sparkConf = new SparkConf().setAppName("DirectKafkaWordCount") .set("spark.executor.memorymessages.map(_.value) val words = lines.f

浏览 9提问于2019-10-21得票数 0

回答已采纳

1回答

StreamingQueryException：‘描述Streams\n===流查询时出错

apache-spark、spark-streaming、aws-glue

我在运行无法连接到Kinesis数据源的胶水流式作业时遇到以下错误：WARNING:root:StreamingQueryException caught.Query ===\nIdentifier: [id = 60exxxxxxxxxxxxxspark-tags_2.11-2.4.0.jar,spark-streaming_2.11-2

浏览 0提问于2020-09-08得票数 0

1回答

如何使用Spark-streaming进行实时日志分析？？(我附加架构镜像)

apache-spark、pyspark、spark-streaming

Spark streaming应用程序实时接收来自大量IoT设备的数据。但它们都是少量的数据。整体流程看起来像是-> Iot -> Kafka(1个主题/所有数据) ->火花流(过滤错误日志) -> DB(保存) ->警报屏幕有没有使用spark或python进行实时日志分析的好方法？

浏览 13提问于2020-11-19得票数 0

回答已采纳

1回答

Spark Streaming -作业以默认spark.streaming.concurrentJobs设置并发运行

apache-spark、spark-streaming

我在Spark Streaming Job中遇到了一种奇怪的行为。我们已使用spark.streaming.concurrentJobs的默认值1。同一流式作业在批处理间隔设置为10分钟的情况下正常运行了一天多。突然，相同的作业开始对所有传入的批处理并发运行，而不是将它们放入队列中。以前有没有人遇到过这种情况？这将是非常有帮助的！

浏览 1提问于2017-04-18得票数 3

1回答

火花结构化流作业不处理阶段并显示为挂起状态

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

我正在运行一个流式应用程序和处理数据从卡夫卡到卡夫卡使用火花。如果我使用的是最新的，那么它的工作，如预期和运行，没有任何问题。num-executors 6 --driver-memory 8G --executor-memory 8G --conf <

浏览 7提问于2022-05-17得票数 0

1回答

如何在pyspark dataframe中将groupby转换为reducebykey？

python、apache-spark、pyspark、apache-spark-sql、spark-dataframe

取而代之的是，我想使用reducebykey。但我是这个领域的新手。请在下面找到我的场景，Step2:输入列总数为15列，其中5列为关键字段，其余为数值。如何用spark方式和map和reducebykey选项来做类似的逻辑。

浏览 0提问于2017-09-21得票数 1

1回答

运行python示例的apache星火流

python、apache-spark、pyspark、spark-streaming

我试图运行示例目录中给出的python火花流作业- Counts words in UTF8 encoded, '\n' delimited text received from the network--jars \ examplesStrea

浏览 4提问于2017-09-20得票数 1

回答已采纳

2回答

为什么Scala编译器在“包火花中的对象SparkConf不能在包org.apache.spark中访问”时失败？

scala、apache-spark、sbt

我无法访问包中的SparkConf。但是我已经导入了import org.apache.spark.SparkConf。import org.apache.spark.rdd.RDD import org.apache.spark.streaming._ val pairs_new = words.map( w => (w,

浏览 0提问于2015-12-05得票数 2

回答已采纳

1回答

为什么StreamingContext以"IllegalArgumentException: requirement :没有注册的输出操作，所以没有执行“来启动IllegalArgumentException？

scala、apache-spark、spark-streaming

请参见下面的错误消息、示例代码、build.sbt和运行程序的命令。代码import org.apache.spark._import org.apache.spark.SparkConfimport org.apache.spark.SparkContext import org.apache.spark.stream

浏览 0提问于2017-12-25得票数 3

1回答

结合使用mqtt和pyspark流

apache-spark、spark-streaming、mqtt

我是spark和mqtt的新手。counts = lines.flatMap(lambda line: line.split(" ")) \ .reduceByKeylambda a, b: a+b) ssc.awaitTermination() 我按照指示安装了蚊子代理(它正在工作)，下载spark-streaming-mqtt

浏览 9提问于2016-09-06得票数 3

1回答

不使用“wordcount.java - example”脚本运行Spark* Kafka示例*

java、apache-kafka、apache-spark

我正在尝试运行Spark中的字数统计示例，它将从Kafka中流式传输数据。。然而，我发现Cloudera Spark发行版与孵化器发行版略有不同。我运行spark shell和从那里运行字数统计示例都没有问题。但是，bin文件夹中没有"run- example“脚本，如示例源代码所示。 * `./bin/run-example org.apache.

浏览 0提问于2014-03-03得票数 4

1回答

使用文件流的火花流字计数不打印结果

scala、apache-spark、spark-streaming、filestream

我使用文件作为Spark流，我想计算流中的单词，但是应用程序什么也不打印，这是我的代码。我在Cloudera环境中使用Scala import org.apache.spark.streaming._ def main(arg

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

回答已采纳

2回答

如何将数据帧中的数据存储在变量中，以用作cassandra中select的参数？

apache-spark、cassandra、apache-spark-sql、spark-structured-streaming、spark-cassandra-connector

我有一个Spark结构的流媒体应用程序。应用程序从kafka接收数据，并且应该使用这些值作为参数来处理来自cassandra数据库的数据。我的问题是，我如何使用输入数据帧(kafka)中的数据，作为cassandra中的"where“参数"select”，而不会出现以下错误： Exception in thread "main" org.apache.spark.sql.AnalysisException: Queries with streaming sour

浏览 25提问于2021-05-24得票数 0

回答已采纳

1回答

火花流的动态分配

apache-spark、spark-streaming、dynamic-allocation、apache-spark-2.0、apache-spark-1.6

我有一个星火流作业运行在我们的集群与其他作业(星火核心作业)。我想使用动态资源分配这些工作，包括星火流。根据下面的JIRA问题，动态分配不支持星火流( 1.6.1版本)。但在2.0.0中是固定的它是否可在星火2.0.0中</

浏览 0提问于2016-12-22得票数 7

1回答

使用Spark* API在Scala中运行独立应用程序示例时出错*

scala、apache-spark

我无法运行spark示例()。sbt运行时出错at org.apache.spark.executor.Executor$$anonfun$org$apache$spark</em

浏览 2提问于2014-02-27得票数 0

1回答

在java中修剪Apache Spark中的特殊字符

scala、apache-spark、hadoop

我尝试删除所有特殊字符，如空格("“)、逗号("，")、分号(";")、冒号(":")、句点(".")、撇号(”‘“)、引号( "”)、感叹号(“！”)、问号(“？”)import org.apache.spark.SparkConfimport org.apache.spark.streamingssc.textF

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

点击加载更多