将JavaPairDStream<String、Integer>转换为JavaPairDStream<Integer、String>、元组中的切换值(Java Spark)
要将JavaPairDStream<String, Integer>转换为JavaPairDStream<Integer, String>，可以使用Spark的转换函数mapToPair()来实现。
以下是完善且全面的答案：
在Java Spark中，JavaPairDStream是用于处理键值对的数据流。要将JavaPairDStream<String, Integer>转换为JavaPairDStream<Integer, String>，可以使用mapToPair()函数来交换键和值的位置。
具体的转换代码如下所示：
JavaPairDStream<String, Integer> originalDStream = ...; // 原始的JavaPairDStream<String, Integer> JavaPairDStream<Integer, String> switchedDStream = originalDStream.mapToPair(tuple -> new Tuple2<>(tuple._2, tuple._1));
在上述代码中，原始的JavaPairDStream<String, Integer>被赋值给了originalDStream变量。然后使用mapToPair()函数对每个元组执行转换操作，创建一个新的Tuple2对象，其中键和值的位置被互换。最后，转换后的JavaPairDStream<Integer, String>保存在switchedDStream变量中。
这种转换在某些场景下非常有用，例如当我们需要根据值进行排序或过滤时，可以方便地使用这种键值对的交换操作。
腾讯云相关产品推荐：
- 对于云计算和大数据处理，腾讯云提供了腾讯云大数据分析平台（DataWorks）。该产品可以帮助用户轻松地进行数据清洗、分析和挖掘工作，提供了丰富的数据处理和计算能力。
- 对于Spark集群的部署和管理，腾讯云提供了腾讯云Spark集群服务。该服务提供了灵活的集群规模调整、自动化管理、可视化监控等功能，使用户能够更好地管理和运行Spark应用程序。
腾讯云大数据分析平台（DataWorks）产品介绍链接地址：https://cloud.tencent.com/product/dc
腾讯云Spark集群服务产品介绍链接地址：https://cloud.tencent.com/product/spark
相关·内容
我是Kafka和spark的初学者。我想通过spark streaming对我从kafka收到的特定主题的数据进行实时处理。我无法使用由createStream函数返回的JavaPairReceiverInputDStream。
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName(
"testwordCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf,
浏览 6提问于2015-02-05得票数 3
我正在尝试写一个简单的应用程序，使用Spark Streaming从Kafka中读入，并持续统计一个单词从一个主题中读入的次数。我在调用非常重要的updateStateByKey方法时遇到了问题，看起来我在泛型方面遇到了问题，但我不确定出了什么问题。
错误：
The method updateStateByKey(Function2<List<Integer>,Optional<S>,Optional<S>>)
in the type JavaPairDStream<String,Integer> is not applicable
浏览 0提问于2016-06-01得票数 1
2回答
我正在处理一个java jar。累加器将流值相加。问题是，每当UI增加或以特定的周期间隔时，我都希望在UI中显示该值。
但是，由于累加器值只能从驱动程序中获得，所以在进程完成其执行之前，我无法访问该值。知道如何定期访问这个值吗？
我的代码如下所示
package com.spark;
import java.util.HashMap;
import java.util.Map;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.fu
浏览 0提问于2014-09-25得票数 2
回答已采纳
我正在尝试学习火花串流，当我的演示主机是"local2“时，这是正常的。但是当我在StandAlone模式下为本地群集启动setMaster时，出现了一个错误: lost an executor 2(已删除)：Unable to create executor in java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME和hadoop.home.dir in unset。 应该注意的是，我在idea中提交了代码 @Component
public final class JavaNetwork
浏览 45提问于2021-08-06得票数 0
回答已采纳
我有一个包含键值对的JavaPairDStream。我需要将它转换成一个HashMap.I，我尝试通过调用"collectAsMap()“函数来对普通的collectAsMap()做同样的操作，但是当我试图在DStream上做同样的操作时，它失败了。
我试图通过使用"JavaPairDStream“函数将" JavaPairRDD”转换为"foreachRDD“函数，然后在JavaPairRDD上使用"collectAsMap()”函数来实现同样的目的。
Map<String,String> value= new HashMap<Str
浏览 3提问于2014-08-10得票数 10
回答已采纳
2回答
我在部署星火流运行的单词计数示例时遇到了问题。我正在尝试部署与Spark示例提供的相同文件，但我希望构建并部署此特定示例，作为独立应用程序。
我的档案是这样的：
package test;
import scala.Tuple2;
import com.google.common.collect.Lists;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Functi
浏览 1提问于2015-07-28得票数 1
回答已采纳
1回答
我已经成功地在Java中构建了一个非常简单的基于的星火流应用程序。
当我将此应用程序提交到本地Spark时，它将等待将一个文件写入给定的目录，并且当我创建该文件时，它将成功地打印单词数。我通过按Ctrl+C来终止应用程序。
现在，我尝试为这个功能创建一个非常基本的单元测试，但是在测试中，我无法打印相同的信息，即单词数。
我遗漏了什么？
下面是单元测试文件，之后我还包含了显示countWords方法的代码片段：
StarterAppTest.java
import com.google.common.io.Files;
import org.apache.spark.streaming.Dura
浏览 4提问于2014-12-08得票数 5
回答已采纳
这是一个有效的代码示例：
JavaPairDStream<String, String> messages = KafkaUtils.createStream(javaStreamingContext, zkQuorum, group, topicMap);
messages.print();
JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
@Override
public String call(T
浏览 1提问于2015-03-27得票数 6
回答已采纳
1回答
我是新手，火花流的概念，并一直困在过去两天试图理解火花流从套接字。我看到火花能够读取传递到插座的块。但是，它不对读取块执行任何操作。
这是星火的代码
package foo;
import java.io.File;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.Jav
浏览 0提问于2014-11-26得票数 0
回答已采纳
1回答
我尝试使用火花流，并希望有一个全局状态对象，可以更新后，每批处理。据我所知，至少有两个选项可供我选择: 1.使用mapWithState，在每一批处理后，火花将自动更新状态。2.使用updateStateByKey函数，其中我必须自己调用更新
对我来说，这两种情况都可以，但我两次都犯了同样的错误。下面是我为这两种情况及其错误编写的示例代码：
Function3<String, Optional<Integer>, State<Integer>, Tuple2<String, Integer>> mappingFunc =
new
浏览 1提问于2016-12-05得票数 1
我有一个带有值的JavaPairDStream<String, Tuple2<Double, Double>>， (ADBE,(1.1,1.2))
(MSFT,(3.2,1.2)) 我想要将元组值相加在一起，并转换为JavaPairDStream<String,Double>作为 (ADBE,(2.3))
(MSFT,(4.4)) 如何在Java Spark streaming中实现这一点？
浏览 12提问于2019-02-16得票数 0
我试着运行一个示例星火流代码。但我知道这个错误：
16/06/02 15:25:42 ERROR streaming.StreamingContext: Error starting the context, marking it as stopped
java.lang.IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute
at scala.Predef$.require(Predef.scala:233)
at org.apache.spark
浏览 3提问于2016-06-03得票数 1
我使用了github仓库中的spark streaming示例程序，并尝试使用kafka和自定义接收器。在这两种情况下，我都会在20-30秒后得到输出。在自定义接收器代码中，我立即获得数据，但输出需要20-30秒。我在单个节点上运行这段代码。
是我做错了什么，还是有优化，我需要执行，还是因为我在单节点上运行。
如果有人可以指导我，这将是一个很大的帮助。
我使用了spark存储库中的代码，以下是代码：
import scala.Tuple2;
import com.google.common.collect.Lists;
import org.apache.spark.SparkConf;
i
浏览 0提问于2014-10-24得票数 4
当使用单例类访问SQlContext时，如下所示在本地模式下工作良好，但是当提交spark master时，它会变成null并抛出空指针异常。如何解决这个问题？在我们的用例中，FlatMapFunction需要查询另一个DStream，返回的结果用于创建新的流。
我对JavaStatefulNetworkWordCount示例进行了扩展，以打印对状态的更改。我需要使用sqlcontext从另一个数据流中的有状态数据流访问rdd，以便创建另一个数据流。如何才能做到这一点？
import java.util.Arrays;
import java.util.List;
import java.ut
浏览 2提问于2015-07-28得票数 0
我曾尝试在spark任务中使用lambda表达式，但它抛出了"java.lang.IllegalArgumentException: Invalid lambda反序列化“异常。当is代码类似于"transform(pRDD->pRDD.map(t->t._2))“时，会抛出此异常。代码片段如下所示。
JavaPairDStream<String,Integer> aggregate = pairRDD.reduceByKey((x,y)->x+y);
JavaDStream<Integer> con = aggregate.trans
浏览 0提问于2015-07-29得票数 7
我一直在尝试用Java实现这一点：
dstream.foreachRDD { rdd =>
rdd.foreachPartition { partitionOfRecords =>
val connection = createNewConnection()
partitionOfRecords.foreach(record => connection.send(record))
connection.close()
}
}
获取Spark文档提供的示例类型。以下代码的工作方式与预期一致：
import scala.Tuple2;
import
浏览 0提问于2015-03-08得票数 3
我想我可以这样做 final JavaPairDstream<String, Integer> y = ... ;
final JavaPairDStream<Integer, String> x =
y.mapToPair(item -> {
return new JavaPairDStream<Integer, String>(item[1], item[0]);
}); 但它给了我一个错误： array required, but scala.Tuple2<java.lan
浏览 24提问于2019-03-15得票数 0
回答已采纳
我试图使用Apache和Spark来流数据，但是我在代码的第24行中发现了一个错误：“无法解决”KafkaUtils“中的”KafkaUtils“方法”。我也尝试过使用KafkaUtils.createDirectStream，但这也不起作用。如何解决此错误？下面是我正在使用的文件
NumbersConsumer.java
import java.util.HashMap;
import java.util.Map;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org
浏览 12提问于2022-11-30得票数 0
我有一个队列，我正在使用Spark Streaming从队列中读取数据。我需要将这些数据(在一些计算之后)写入2个不同的表(两个表的计算是不同的)。我注意到我的代码只执行第一个JavaPairDStream，而不执行另一个。我在两个PairStreams上都有输出操作。
我还注意到，如果队列中的数据很小-比方说只有一条消息，那么它会被插入到两个表中，但当队列中的数据很大时就不是这样了
我的代码如下所示-
public static void main(String ar[]) {
JavaReceiverInputDStream<String> receiverStream
浏览 0提问于2016-09-08得票数 0
1回答
我正在尝试使用星火流，但我只停留在第一个例子：
import java.util.Arrays;
import org.apache.spark.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.*;
import org.apache.spark.streaming.api.java.*;
import scala.Tuple2;
public class NetworkWordCount {
public static void main(String[] args
浏览 0提问于2021-02-15得票数 0
回答已采纳
我正在学习Apache流，并尝试从JavaPairInputDStream生成JavaStreamingContext。下面是我的代码：
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.sp
浏览 2提问于2017-12-18得票数 0
回答已采纳
我正在运行一个程序，它使用Apache从Apache集群获取数据，并将数据放在Hadoop文件中。我的计划如下：
public final class SparkKafkaConsumer {
public static void main(String[] args) {
SparkConf sparkConf = new SparkConf().setAppName("JavaKafkaWordCount");
JavaStreamingContext jssc = new JavaStreamingContext(sparkConf
浏览 3提问于2016-07-21得票数 1
回答已采纳
我在Scala中有一个通用方法
def updateStateByKey[S](updateFunc: JFunction2[JList[V], Optional[S],
Optional[S]]) : JavaPairDStream[K, S] = { ... }
当我在Java中调用它时，这两种方法都不会编译：
1
JavaPairDStream<String, Integer> stateDstream =
pairs.<Integer>updateStateByKey(...);
2
JavaPairDStream<String, Integer>
浏览 0提问于2015-03-18得票数 1
回答已采纳
我使用的是spark 1.5.0和java 7。
输入来自type字段的不同json对象形式的kafka。例：
{'type': 'alpha', ...}
{'type': 'beta', ...}
...
我正在根据每个事件类型的计数从这个输入数据创建一个JavaPairDStream<String, Integer>。
我想把这些数据存储到redis。我该怎么做呢？
浏览 2提问于2015-11-19得票数 0
回答已采纳
我正在尝试使用Java而不是scala从apache流连接到Kafka。
我收到以下错误：
由: org.apache.spark.util.SystemClock.() at java.lang.Class.getConstructor0(Class.java:3082) at java.lang.Class.newInstance(Class.java:412)引起的
创建JavaStreamingContext对象时会弹出错误。
这是我的密码：
JavaStreamingContext jssc = new JavaStreamingContext("local[2]&#
浏览 3提问于2017-06-28得票数 0
回答已采纳
1回答
我正在使用Spark2.1.1Scala2.11.8和Java 8
主要Java火花类：
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spa
浏览 0提问于2018-03-17得票数 1
回答已采纳
我尝试使用transform将JavaPairRDD连接到JavaPairDStream中的RDD，但得到以下错误：
incompatible types: no instance(s) of type variable(s) W exist so that org.apache.spark.api.java.JavaPairRDD<java.lang.String,scala.Tuple2<LogType,W>> conforms to org.apache.spark.api.java.JavaRDD<U>
下面是我的代码：
// Load full t
浏览 0提问于2016-03-02得票数 0
1回答
用多键Java减少火花流
、、、
我对星火流非常陌生，我正试图弄清楚如何处理这个问题，因为我发现了很多关于单个(K，V)对的例子，但是还有更多的例子。我希望得到一些帮助，以便找到使用Spark的Java转换的最佳方法。
让我简单地描述一下情况，
目标是在时间窗口内获得一组元素的错误比率。
考虑到以下输入，
(A, Error)
(B, Success)
(B, Error)
(B, Success)
(C, Success)
(C, Error)
它将按元素聚合，然后按状态(Element, (Number of Success, Number of Error))聚合。在这种情况下，转换的结果是，
(A, (0,1))
(B
浏览 1提问于2016-11-21得票数 1
回答已采纳
我使用socket流作为输入流，然后执行map和reduce。
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(2));
JavaReceiverInputDStream<String> logMessage = jssc.socketTextStream("loca
浏览 0提问于2016-06-03得票数 0
我在Ubuntu中以独立模式运行Spark 2.4.3。我正在使用Maven创建JAR文件。下面是我试图运行的代码，它的目的是从Twitter流式传输数据。一旦Spark启动，Spark master将在127.0.1.1:7077。正在使用的java版本是1.8。
package SparkTwitter.SparkJavaTwitter;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Voi
浏览 1提问于2019-11-11得票数 0
1回答
我试图在每个RDD中添加max和min，在它的元组的星星之火dstream..each中。我编写了以下代码，但无法理解如何传递参数min和max。有人能建议一种方法来做这个转变吗？我尝试了以下几点：
JavaPairDStream<Tuple2<Long, Integer>, Tuple3<Integer,Long,Long>> sortedtsStream = transformedMaxMintsStream.transformToPair(new Sort2());
class MinMax implements Function<JavaPa
浏览 1提问于2015-06-17得票数 1
1回答
我是Spark的新手，我正在学习一个教程，其中包含几个字段的一行用Scala解析，scala的代码如下所示：
val pass = lines.map(_.split(",")).
map(pass=>(pass(15),pass(7).toInt)).
reduceByKey(_+_)
其中pass是从socketTextStream (其SparkStreams)获取的数据。我是星火公司的新手，我想使用Java来获得同样的结果。我已经用：
JavaReceiverInputDStream<String> lines = jssc.socketTextStr
浏览 2提问于2015-12-09得票数 0
回答已采纳
1回答
火花流变换函数
、、
我在火花流的transform函数中有编译错误。特别是，似乎没有最后确定DStream变量或类似的内容。我抄袭了amplab的教程，所以有点困惑.
这是代码，问题是在transform函数的末尾。
以下是错误：
[ERROR] /home/nipun/ngla-stable/online/src/main/java/org/necla/ngla/spark_streaming/Type4ViolationChecker.java:[120,63] error:
no suitable method found for transform(<anonymous Function<J
浏览 5提问于2015-06-15得票数 0
回答已采纳
我是Spark API的新手，也是一路学习的新手。
我在hadoop目录中有多个文件，我正在使用wholeTextFiles来创建JavaPairRDD<String, String>。程序是用Java编写的。
我的要求是处理目录中的文件列表并实现以下输出：
文件路径，字
文件路径，字
文件路径，字
..。
这基本上是与对应的文件名(或路径)配对为<String, String>的文件的word内容。
但是，不允许从tuple2转换到Iterable (在运行时失败)：
JavaSparkContext sc = new JavaSparkContext(new
浏览 0提问于2018-08-07得票数 0
回答已采纳
1回答
我正在开发一个Spark流应用程序，在这个应用程序中，我需要使用Python中的两个服务器的输入流，每个服务器每秒向Spark上下文发送一个JSON消息。
我的问题是，如果我只对一个流执行操作，一切都会正常工作。但是，如果我有来自不同服务器的两个流，那么火花在打印任何东西之前就会冻结，只有当两个服务器都发送了它们必须发送的所有JSON消息(当它检测到'socketTextStream没有接收数据时‘)，它才能再次开始工作。
这是我的代码：
JavaReceiverInputDStream<String> streamData1 = ssc.socketTextStr
浏览 6提问于2016-05-09得票数 1
回答已采纳
我正在试验星火作业，它从Kafka中流数据并生产给Cassandra。
我正在使用的示例在给定的时间间隔内获取大量单词，并将单词计数发布到Cassandra。我还试图同时发布时间戳和单词及其计数。
到目前为止，我的情况如下：
JavaPairReceiverInputDStream<String, String> messages =
KafkaUtils.createStream(jssc, zkQuorum, groupId, topicMap);
JavaDStream<String> lines = messages.map(Tuple
浏览 1提问于2017-05-11得票数 0
回答已采纳
我是Spark和Spark流媒体的新手，刚刚开始我的第一步。似乎我真的不知道如何让它工作。
我有一个包含Tuple2<Integer,List<Integer>>的JavaPairDstream。第一个Integer是传感器id，列表是传感器读数。我需要提取列表元素，以便计算传感器读数的平均值。
我试着使用foreachRDD，但它对我不起作用。有人可以发布一个代码片段，伪代码或一些如何做到这一点的指导吗？Java 8代码也会受到欢迎。
谢谢
盖伊
浏览 3提问于2017-04-27得票数 0
我已经扩展了现有的WordCount示例，并在Apache的官方站点上解释了这个例子。延期如下：
根据元组的存在性对元组进行排序。例如:现有的未排序顺序：(nec,8)，(eu,11)，(sit,7)我想要的排序顺序：(eu,11)，(nec,8)，(sit,7)
从排序列表中拿出前三名。排序做得很好，但是top()不在JavaPairRDD上工作。让我粘贴我的代码。其他方法是相同的，所以我在这里写我的主要方法：
公共静态空主(String[] args) { if (args.length < 1) { System.err .println(“请提供输入文件完整路径
浏览 3提问于2015-06-23得票数 0
我运行了一个用Java编写的Spark流程序来读取Kafka的数据，但是由于我使用scala或java的版本很低，所以我试图找出这个错误。我使用了JDK版本15，但是仍然有这个错误，有人能帮我解决这个错误吗？谢谢。
--这是我运行项目时的终端：
Exception in thread "main" java.lang.NoClassDefFoundError: scala/Product$class
at org.apache.spark.streaming.kafka010.PreferConsistent$.<init>(LocationStra
浏览 20提问于2021-05-31得票数 3
回答已采纳
我正试图整合卡夫卡消息代理，激发并面对一个Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/spark/streaming/kafka010/LocationStrategies的问题。
下面是java火花代码
package com.test.spell;
import java.util.Arrays;
/**
* Hello world!
*
*/
import java.util.HashMap;
import java.util.HashSet;
import java.
浏览 0提问于2018-06-29得票数 0
回答已采纳
4回答
我正在尝试将Spark流输出保存到HDFS上的文件中。现在，它没有保存任何文件。
下面是我的代码：
StreamingExamples.setStreamingLogLevels();
SparkConf sparkConf = new SparkConf().setAppName("MyTestCOunt");
JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, new Duration(1000));
JavaReceiverInputDStream<String> lines =
浏览 0提问于2014-07-28得票数 2
我试图整合星火和卡夫卡，因为它消耗了卡夫卡的信息。我也有生产者代码来发送关于“临时”主题的信息。另外，我正在使用卡夫卡的控制台制作人来制作关于“临时”主题的信息。
我已经创建了下面的代码来使用来自同一个“临时”主题的消息，但是它也不会接收单个消息。
计划：
import java.util.Arrays;
import java.util.Map;
import java.util.HashMap;
import static org.apache.commons.lang3.StringUtils.SPACE;
import org.apache.spark.SparkConf;
impo
浏览 3提问于2017-08-31得票数 1
我已经将一个广播数组列表定义为公共静态，该数组列表(数组列表名称："qList")在作业处理程序方法中启动新作业时填充了新值，然后在DStream lambda闭包中使用该数组列表，但在火花集群上运行时，失败的作业会显示消息"Null指针异常“：
由: org.apache.spark.SparkException:因阶段失败而中止的作业:阶段17.0的任务1失败4次，最近的失败: 17.0阶段的任务1.3丢失(TID 40,192.168.1.97，执行者0)：java.lang.NullPointerException at QProcessing.lambda$3
浏览 1提问于2021-03-27得票数 0
我正在尝试使用Spark Streaming 1.2.0。在某种程度上，我将流数据按键分组，然后对它们进行一些操作。
以下是测试代码的一段：
...
JavaPairDStream<Integer, Iterable<Integer>> grouped = mapped.groupByKey();
JavaPairDStream<Integer, Integer> results = grouped.mapToPair(
new PairFunction<Tuple2<Integer, Iterable&l
浏览 0提问于2015-01-22得票数 2
经过很长一段时间的搜寻，它是无法帮助的，我需要问你！我想做一个简单的单词计数的标签，从Tweets与Apache。应用程序从Kafka获取哈希标签，直到reduceByKey函数才能正常工作。(我知道推特和星火之间有直接的联系)
如果没有这个函数，结果如下：
-------------------------------------------
Time: 1483986210000 ms
-------------------------------------------
(Presse,1)
(Trump,1)
(TheResistanceGQ,1)
(MerylStreep,1)
(t
浏览 3提问于2017-01-09得票数 1
回答已采纳
1回答
我计划实现星火流，以便实时评估规则，而不是从数据库中执行。
我在卡夫卡的话题中吸收了一些数据。所摄取的数据是关于用户和他的行为。我有这样的规则:在10分钟内识别超过100次事务的用户，在最后5分钟内识别执行0次事务的用户。
时间间隔(窗口)和规则在某种程度上是不同的。我有大约100条这样的规则。当我们在创建流时指定窗口时，我们需要创建与规则数量一样多的流。我不确定这是一个正确的方法和星火流适合这个用例。因为我是新火花，我正在寻找投入，以有效地解决这一问题。
请查找下面的示例代码和规则。这些规则将由分析师配置，可以有50条规则，时间周期可以根据分析师的需求变化，从1分钟到1小时不等。如果我有10
浏览 0提问于2016-01-13得票数 0
我已经使用Spark (1.3.0)、Scala (2.10.4)和Spark avro (1.0.0)将Spark数据帧保存到Avro文件中。
def getMatchingLine(line: String) : Option[Row] = {
val regex = "^.*&50=(\\w+)&.*&62-\\d=8&63-\\d=(\\w+)&.*timestamp=(\\d+).*$".r
line match {
case regex(a, b, c) => Some(Row(a, b,
浏览 0提问于2016-03-13得票数 4
回答已采纳
当我试图执行我的kafka spark项目时。我得到了下面的错误：
Exception in thread "main" java.lang.NoClassDefFoundError: org/spark_project/guava/cache/CacheLoader
at org.apache.spark.SparkConf.loadFromSystemProperties(SparkConf.scala:73)
at org.apache.spark.SparkConf.<init>(SparkConf.scala:68)
at org.
浏览 60提问于2017-02-22得票数 1
回答已采纳
我创建了一个有5个分区的Kafka主题。我正在使用createStream接收器API，如下所示。但不知何故，只有一个接收者正在获取输入数据。其余的接收者不进行任何处理。你能帮忙吗？
JavaPairDStream<String, String> messages = null;
if(sparkStreamCount > 0){
// We create an input DStream for each partition of the topic, unify those streams, and then re
浏览 1提问于2016-09-14得票数 1
我是星火的新手。我正在试验使用Spark2.1版本作为CEP的目的。在最后2分钟内检测缺少的事件。我将接收到的输入转换为JavaDSStream的输入事件，然后在inputEvents上执行reducebykeyandWindow并执行spark。
JavaPairDStream<String, Long> reduceWindowed = inputEvents.reduceByKeyAndWindow(new MaxTimeFuntion(),
Durations.seconds(124), new Duration(2000));
re
浏览 0提问于2017-04-10得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
