我正在使用spark streaming，我开发了以下spark streaming应用程序：
它们分别从Kafka接收器(RDD1)和HTTP请求(RDD2)创建一个DStream。我的问题是，我只想使用RDD1中的第一个元素，并在我的RDD2中使用它，而这段代码在spark streaming (.first())中不起作用，我如何才能在spark streaming 1.6中获得相同的结果
浏览 9提问于2017-06-29得票数 0
当我使用spark streaming从Cassandra读取数据时，我遇到了一个问题。, 3)我的代码如下所示，希望您的回复。import org.apache.spark.streaming._
import com.datastax.spark.conn
浏览 2提问于2015-09-08得票数 10
回答已采纳
1回答
$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:358) at org.apache.spark.sql.executio
浏览 184提问于2021-07-11得票数 0
回答已采纳
:112) at org.apache.spark.rdd.RDD.collect$org$apache$spark$sql$Dataset$$collect$1.apply(Dataset.scala:2197) at org.apache.spark.sq
浏览 2提问于2017-06-06得票数 0
回答已采纳
2回答
(“，3)，(”猫“，2)，(”喜欢“，1)，(”培根“，2)，(”我的“，1)，(”吃“，1)下载和untarspark-1.5.2-bin-hadoop2.6bin/pyspark --packages org.apache.spark:spark-streaming-kafka_2.10:1.5.2基于中的示例。from pyspark.streaming import StreamingContext
from
浏览 3提问于2015-11-15得票数 3
函数时没有任何效果：import sysfrom pyspark.streaming.kafka import KafkaUtilsfrom kafka importproducer = SimpleProducer(kafka)
浏览 4提问于2016-05-20得票数 8
回答已采纳
2回答
) + '\n').encode()
tcp_connection.send(send_batch) # send the payload to Spark
浏览 1提问于2021-12-04得票数 0
回答已采纳
($"timestamp", "10 minutes", "5 minutes"), .agg(count("value") as "kafka.count",collect_setcounter1 = event_stream .count()
// .agg(count("value") as &q
浏览 5提问于2018-07-18得票数 3
org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:944) at org.apache.spark.rdd.RDD.withScope(RDD.scala:370)
at org.apache.spark.rdd.RDD.collectat org.apache.spark.sql.D
浏览 0提问于2018-08-24得票数 1
回答已采纳
$anonfun$collect$1(Dataset.scala:2782) at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3364)
at org.apache.spark.sql.Dataset.collect(Dataset.sca
浏览 22提问于2020-09-15得票数 0
回答已采纳
1回答
我正在使用Shell (Scala2.10和星火流org.apache.spark:spark-streaming-kafka-0-10_2.10:2.0.1)来测试星火/卡夫卡消费者：import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.Consume
浏览 1提问于2016-11-15得票数 0
回答已采纳
以下代码的工作方式与预期一致：import com.google.common.collect.Lists;
import org.apache.spark.streaming.Durations;import org.apache.spark
浏览 0提问于2015-03-08得票数 3
package com.scala.sparkStreaming
import org.apache.spark.streaming._
at org.apache.spark.streaming.StreamingContext.validate(StreamingContext.scala:517)
at org.apache.spark.st
浏览 2提问于2020-05-22得票数 1
我在用apache spark注入twitter数据。如何在数据流statuses上执行此操作？除了print()之外，我似乎不能让任何东西工作。import org.apache.spark._import org.apache.spark.streaming._import org.apache.spark.streaming.twitter._
import org.apache.<e
浏览 0提问于2014-07-16得票数 7
回答已采纳
root_.kafka.message.MessageAndMetadataimport org.apache.spark._import org.apache.spark.streaming.kafka._
kafkaStream.foreachRDD(r
浏览 2提问于2015-12-02得票数 1
回答已采纳
我是spark的新手，试着写一些基于spark和spark streaming的示例代码。到目前为止，我已经在spark中实现了排序函数，代码如下： val conf = new SparkConfval finalResult = result.collect() printlnArray(
浏览 1提问于2015-01-06得票数 6
我正在做一个数据分析项目，在这个项目中，我从CSV文件中读取数据，在Kafka主题上遍历该文件，并使用Spark Streaming来使用该Kafka主题数据。我在一个单一的项目中使用的所有组件。现在，在使用Spark Streaming消费数据后，我必须对其进行一些计算，我必须将数据保存到elastic search中，并且我必须将该数据发送到另一个主题。所以我从Spark Streaming开始做这些事情(将数据保存到elastic中，并将数据发送到topic)。2)我在一个项目中使
浏览 0提问于2019-06-04得票数 0
1回答
我想要做的是从Couchbase视图检索所有相关文档，并从Spark流中获得给定的Id。import org.apache.spark.streaming._ // get
浏览 3提问于2016-02-03得票数 0
回答已采纳
我正在尝试创建一个spark consumer API来在我的消费者代码中接收来自Kafka.But的数据，我无法为这两个类添加jar/dependency :导入org.apache.spark.streaming.scheduler.ReceiverLauncher；导入org.apache.spark.streaming.Scheduler；package kafkatest2o
浏览 1提问于2017-09-27得票数 0
我有一个Spark Streaming应用程序，可以从Kafka中读取记录中的配置单元表名，例如table1。table2..table3..等。// code to write to kafka当我执行此命令时，我在sparkSession.sql调用中得到一个NullPointerException，如下所示： at org.apache.spark.s
浏览 0提问于2019-02-25得票数 1
云服务器
ICP备案
云直播
即时通信 IM
对象存储
腾讯云开发者
