我正在学习spark streaming，并从一些生产日志中找出一些散列标签。在一些示例中，我发现了以下代码： val words = statuses.flatMap(line => line.split(" "))
val tagCounts = tagKe
浏览 29提问于2019-03-17得票数 0
我想利用Spark的结构化流功能，但我需要在一个窗口内做一个reduce by键。基本火花流支持一种称为reduceByKeyAndWindow的方法。我对Spark Structured Streaming文档进行了大量的搜索，但没有找到任何与Spark Streaming的reduceByKeyAndWindow相当的文档。有没有什么结构化流式操作的组合可以让
浏览 0提问于2018-04-14得票数 2
1回答
火花流窗口操作
、、、
下面是一个简单的代码，用于获取窗口大小为30秒和幻灯片大小为10秒的单词计数。import org.apache.spark.SparkConfimport org.apache.spark.streaming.StreamingContext._import org.ap
浏览 0提问于2014-07-22得票数 6
回答已采纳
1回答
我正在尝试使用Spark Streaming将Kafka输入索引到elasticsearch中。kafka中的消息是这样的：我想在Spark Streaming中定义结构，以便在elasticsearch中索引此消息：我读过有关RDD转换<e
浏览 1提问于2016-02-10得票数 1
我是星火的新手，所以请指点。你可以从那里查看。但当我执行
./run-example org.apache.spark.streaming.examples.TwitterPopularTags local[2]然后它将向我展示身份验证失败的问题..。
浏览 0提问于2013-10-30得票数 2
回答已采纳
当我使用spark streaming时，我不太理解转换操作，下面是我的代码： ssc .map((_, 1))
.reduceByKeyAndWindow
浏览 4提问于2017-06-20得票数 0
我正在使用Apache Spark streaming对我的web服务API日志进行一些实时处理。源流只是一系列带有返回码的API调用。我的Spark应用程序主要是对原始API调用日志进行聚合，计算有多少API返回特定的HTTP代码。val level2 = level1.reduceByKeyAndW
浏览 1提问于2015-04-30得票数 2
当我运行下面的代码时，我会得到这个错误import org.apache.spark.streaming.twitter._唯一与日志记录有关的部分是当我调用sc.setLogLevel("ERROR")时。但我已经试过移除这个错误了。name := "SparkStreaming"
versi
浏览 3提问于2021-04-28得票数 0
回答已采纳
1回答
我正在使用Kafka和Spark来统计twitter上最受欢迎的标签。这就是我想要运行的scala对象：import org.apache.spark.streaming.kafka._
import org.apache.spark.streaming.{ Secon
浏览 10提问于2018-02-12得票数 0
回答已采纳
我对Spark有相当多的经验。然而，我是第一次学习Spark streaming，我发现它很难理解。很难理解应用程序的流程。例如，对于以下站点上给出的示例
val words = statuses.flatMap(status => status.split(" "))
val hashtags = words.filter(word=> word.startsW
浏览 0提问于2015-08-18得票数 1
1回答
在星火流中，设置StreamingContext以将检查点设置为可靠的数据存储(S3、HDFS、.)是可能的(如果要使用有状态操作，也是强制性的)。(和)：
另一方面，可以通过在每个DataStream上调用checkpoint(timeInterval)来
浏览 4提问于2015-12-31得票数 13
回答已采纳
我有一个流应用程序，可以获取数据流并在数据流上运行sql操作，然后将其转储到文件中{spark.read.json(rdd)dstream.foreachRDD { rdd =>
{val df = spark.read.json(rdd有没有办法以某种方式将计算结果写入内存，并将其用作计算
浏览 0提问于2017-11-20得票数 0
import org.apache.spark.streaming.StreamingContextimport org.apache.spark.streaming.Secondsval searchPair=hottestStream.map(_.split("")(1)).map(item=>(item,1))
val hottestDStream=searchPair.reduceByKey
浏览 5提问于2017-05-27得票数 4
回答已采纳
我试着用Spark Streaming从Twitter上流式传输数据。但是import twitter4j.auth._import org.apache.spark.streaming{Seconds,StreamingContext}import org.apac
浏览 4提问于2017-06-15得票数 0
3回答
我有Flume Avro接收器和读取接收器的SparkStreaming程序。CDH 5.1、Flume 1.5.0、Spark 1.0，使用Scala作为Spark上的程序语言
我能够制作Spark示例并计算Flume Avro事件。有没有人有一个使用Scala的例子？
浏览 0提问于2014-08-25得票数 0
下面这段代码是我在Spark Streaming上使用的Twitter流媒体应用的一部分：我在我的F:\Software=***
oauth.a
浏览 49提问于2020-05-08得票数 0
TL;DR:如何在Spark Mongo连接器中注册编解码器？libraryDependencies ++= Seq( "org.apac
浏览 1提问于2017-01-09得票数 1
是否可以限制Kafka消费者返回Spark Streaming的批量大小？
我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。
浏览 0提问于2016-10-11得票数 18
回答已采纳
我在尝试运行字数统计示例时遇到以下错误。
不确定如何继续。下面是我正在运行的命令和错误。/opt/spark/bin/spark-submit --jars spark-streaming_2.10-2.0.0.jar test_kafka.py broker.txt "localhostkvs = KafkaUtils.createDirectStream(ssc, [topic], {"m
浏览 9提问于2016-08-29得票数 3
我对Spark和Scala非常陌生，我正在使用ReduceByKeyAndWindows来计算kafka消息中的单词，因为我需要使用窗口功能。我的应用程序的目的是在检测到卡夫卡在特定时间包含特定单词的"x“次消息时发出警报。然后，从开始重新启动。import org.apache.spark
浏览 4提问于2016-03-25得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
