我正在使用Spark Kafka连接器从Kafka集群获取数据。从中，我将以JavaDStream<String>的形式获取数据。在EventLog是Java bean的情况下，如何获得JavaDStream<EventLog>形式的数据？public static JavaDStream<EventLog> fetchAndValidateData(String
浏览 21提问于2016-08-11得票数 0
回答已采纳
对于一个类项目，我需要一个Spark Java程序作为Kafka消费者来监听，并将Kafka主题的所有接收到的消息写入一个文件(例如"/user/zaydh/my_text_file.txt")。我可以将消息作为JavaPairReceiverInputDStream对象接收；我还可以将其转换为JavaDStream<String> (来自Spark Kafka示例)。但是
浏览 0提问于2015-10-22得票数 0
SparkStreamingContext是在驱动程序上创建的，无法序列化。所以，我们不能在workers上访问这个对象。现在，为了利用Spark的分布式特性，我们将在workers上处理数据/流。因此，直接从工作进程保存数据是有意义的，而不是将所有数据收集到驱动程序，这可能会产生OOM。
我有这样的场景，我需要应用一些业务逻辑，最后需要将数据存储到cassandra。那么，我该怎么做呢？
浏览 12提问于2016-09-09得票数 1
1回答
闪回的是：我能够创建与cassandra的连接，但是数据没有被插入到cassandra表中。</groupId> <version>1.1.
浏览 0提问于2014-12-09得票数 0
回答已采纳
我已经创建了一个JavaPairInputDStream，并尝试将消耗的数据保存到Cassandra表中。但是面临着问题，并且不确定如何从代码开始：package com.test.anna.KafkaSpark;
import org.apache.spark.streamin
浏览 0提问于2017-10-05得票数 0
1回答
我有像下面这样的代码从Kafka中读取并保存到Elasticsearch，我正在使用spark streaming： jssc.start();
我的问题是，我需要根据主题的名称使&quo
浏览 5提问于2018-02-04得票数 0
回答已采纳
我使用星火消费卡夫卡的数据，并保存在卡桑德拉。我的程序是用Java编写的。我正在使用spark-streaming-kafka_2.10:1.6.2库来完成这个任务。我的代码是：JavaStreamingContext jssc = new JavaStreamingContextStringDecoder.class, EventLogDecoder
浏览 1提问于2016-08-26得票数 11
回答已采纳
1回答
我尝试运行java kafka consumer for spark，无论我做什么，我都会得到异常。在我看到的异常(ConsumerStrategy.scala:85)中，为什么这里写着scala？我的库中是否有冲突？;import org.apache.spark.streaming.Durations;
import org.apache.spark.stre
浏览 59提问于2020-04-08得票数 1
回答已采纳
我从linux终端的流中获取一些条目，将它们分配为lines，并将它们分解为words。但我不想把它们打印出来，而是把它们保存给卡桑德拉。我有一个名为ks的密钥空间，其中有一个表名为record。代码：import org.apache.spark.SparkConf;
import org.apache.spark</em
浏览 2提问于2016-05-12得票数 1
回答已采纳
我正在运行一个程序，它使用Apache从Apache集群获取数据，并将数据放在Hadoop文件中。我的计划如下： public static void main(String[] args) {:2181", "NameConsumer", topicMap);
JavaDStream<String> lines = messages.map(new Function<Tu
浏览 3提问于2016-07-21得票数 1
回答已采纳
我没有从使用Kafka直接流的队列中获得任何数据。在我的代码中，我放入了System.out.println()，这条语句没有运行，这意味着我没有从该主题获得任何数据。我非常确定队列中的数据是可用的，因为没有进入控制台。有没有人能提个建议？下面是我的Java代码，
SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWord
浏览 2提问于2016-11-23得票数 1
如何在Spark Structured Streaming中获取Kafka header字段(在Kafka 0.11+中引入)？我看到headers实现是在Spark 3.0中添加的，但在2.4.5中没有。我看到默认情况下spark-sql-kafka-0-10使用的是kafka-client 2.0。如果不能使用Spa
浏览 0提问于2020-03-09得票数 0
我想把Kafka + Cassandra连接到Spark 1.5.1。这些库的版本： "org.apache.spark" % "spark-streaming_2.10" % "1.5.1",
"org.apache.spark" % "spark</e
浏览 2提问于2015-11-07得票数 1
Spark version:3.00Cassandra::3.11.4 .builder \
.config("spark.jars","spark-streaming-kafka-0-10_2.12-3.0.0.jar,spa
浏览 27提问于2020-06-27得票数 2
回答已采纳
1回答
我正在开发一个PySpark应用程序，它使用星火流从Kafka提取数据。我的消息是字符串，我想在Scala代码中调用一个方法，向它传递一个DStream[String]实例。但是，我无法在Scala代码中接收到适当的JVM字符串。在我看来，Python字符串不是转换成Java字符串，而是序列化的。我正在PySpark
浏览 4提问于2016-09-12得票数 5
回答已采纳
我用的是火花流。根据火花编程指南(参见)，命名累加器将显示在WebUI中，如下所示：
ssc.sparkContext.sc().register(accumulator, "my accumulator");
我使用的是Spark 2.0.0。
浏览 2提问于2015-04-24得票数 12
我正在试验星火作业，它从Kafka中流数据并生产给Cassandra。JavaDStreamwords.mapToPair(s -> new Tuple2<&
浏览 1提问于2017-05-11得票数 0
回答已采纳
Spark从kafka获取数据。我想通过spark-streaming限制记录消耗。关于卡夫卡的数据非常庞大。我使用spark.streaming.kafka.maxRatePerPartition=1属性来限制spark中的记录。但是仍然在5分钟内，我收到了13400条消息。我的spark程序每5分钟不能处理超过1000条消息。Kafka topic有3个分区。
浏览 17提问于2018-02-06得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
