我有一个带有值的JavaPairDStream<String, Tuple2<Double, Double>>， (ADBE,(1.1,1.2))(MSFT,(4.4)) 如何在Java Spark streaming中实现这一点？
浏览 12提问于2019-02-16得票数 0
1回答
用多键Java减少火花流
、、、
让我简单地描述一下情况，(B, 33.3%)据我所知，结果将由reduceByKeyAndWindow()函数提供，例如，根据申请的相反流程，我的问题是，
如何在JavaPairDStream上定义具有多个值
浏览 1提问于2016-11-21得票数 1
回答已采纳
我了解到，在使用DStreams作为print()时，只承认了一些输出操作。这是一段代码 JavaPairDStream<String<String, Integer> lines = messages.mapToPair((x)->(new Tuple2<String, Integer>(x._2, 1))).reduceByKey(sumFunc);
浏览 6提问于2017-06-16得票数 2
回答已采纳
1回答
在我的应用程序中，我正在为多个定制接收器(为不同的UDP数据套接字使用2个接收器，为HTTP数据使用1个接收器)使用Spark流。接收器的转换没有任何共同的资源。reduce.foreachRDD(rdd -> DebugUtil.saveTestDataToDisk(rdd, "reduce" + "-" + suff));
JavaPairDStream<String, NginxRaw> uidRaw
浏览 1提问于2017-12-29得票数 4
<String,ArrayList<String>> history = history_dstream.mapToPair(r -> {然后按下面的方式进行连接，然后过滤掉那些没有连接键的记录，并通过与Data_1联合使用，将它们保存在历史记录中，以便在下一批中使用。Data_1 = Data_1.union(history);
JavaPairDS
浏览 2提问于2017-06-08得票数 1
回答已采纳
我尝试使用transform将JavaPairRDD连接到JavaPairDStream中的RDD，但得到以下错误： flumeEvent -> { LogType
浏览 0提问于2016-03-02得票数 0
我有一个在本地模式下运行的Spark流程序，在这个模式中，我从一个TCP套接字连接接收JSON消息，每个批处理间隔几个。这些消息中的每一条都有一个ID，用于创建一个键/值JavaPairDStream，因此在我的JavaDStream中的每个RDD分区中，都有一个键/值对，每个分区只有一个消息。StorageLevels.MEMORY_AND_DISK_SER);
JavaDStream<String>
浏览 2提问于2016-06-19得票数 2
回答已采纳
1回答
= null); .mapToPair(s -> newTuple2<String, CDR>(s
JavaPairDStream<String, Tuple2<CDR, List<<String,Tuple2<CDR,List&
浏览 1提问于2016-03-30得票数 0
1回答
String> call(String s) { return Arrays.asList(s.split(",")); } });
JavaPairDStream<String, Integer> pairs = words.map( public Tuple2
浏览 2提问于2015-12-09得票数 0
回答已采纳
我曾尝试在spark任务中使用lambda表达式，但它抛出了"java.lang.IllegalArgumentException: Invalid lambda反序列化“异常。JavaPairDStream<String,Integer> aggregate = pairRDD.reduceByKey((x,y)->x+y);
JavaPairDStream</
浏览 0提问于2015-07-29得票数 7
我编写了一个用于接收来自textSocketStream的数据的星火程序，我正在计算一个温度值的平均值。当我停止在~1 min之后将数据发送到我的Spark集群时，窗口的时间(即1h )的平均值不应该改变，所以有关于59 min over的内容，在那里什么都不应该改变！现在我发现了一个问题:对我来说，这个数据量是正确的:窗口DStream中的100个条目，但是值的计算和(以及通过avg = sum / count计算的平均值)在几个不同的平均值之间波动。在这里，控制台输出片段(在</em
浏览 1提问于2016-09-02得票数 1
1回答
使用这个ID作为密钥，我可以执行一个MapPartitionsToPair，从而创建一个JavaPairDStream，其中包含一个键/值对的RDD，每个分区一个键值对(因此，如果我收到5个JSON消息，就会得到一个具有5个分区的RDD，每个分区都以消息的ID作为键，JSON消息本身作为值)。我现在要做的是，将所有具有相同键的值分组到同一个分区中。ssc.socketTextStream(args[0], Integer.parseInt(args[1]),
Storag
浏览 3提问于2016-06-19得票数 4
回答已采纳
1回答
JavaInputDstream API有一个从特定偏移量开始的选项，但是我需要它，因为JavaPairInputDstream是解决方法，我们可以使用它来获取JavaPairInputDstream，因为在API中它是不可用的。
浏览 3提问于2017-11-29得票数 0
我试图在星火流中执行一个MapPartitionsToPair，但是它总是返回错误: NoSuchElementException: next on空迭代器。这是我的代码： @Override
pub
浏览 3提问于2016-06-17得票数 0
回答已采纳
在某种程度上，我将流数据按键分组，然后对它们进行一些操作。...JavaPairDStream<Integer, Iterable<Integer>> grouped = mapped.groupByKey(); @Override
public Tuple2<Integer, Integer>
浏览 0提问于2015-01-22得票数 2
我在Spark Streaming代码中有这个函数，它可以将推文分成单独的单词 .flatMapJavaPairDStream<String, String> wordTweets = statuses.flatMapToPair( public Iterabl
浏览 0提问于2015-04-23得票数 0
0回答
JavaPairDStream<String, Integer> wordCounts = words.mapToPair( @Override returnnew Tuple2<>(s, 1); }).reduce
浏览 2提问于2017-06-10得票数 2
回答已采纳
createStream(jssc, "localhost:2181", "kafkasparkconsumer",下面的代码给出了一个错误： new PairFunction<String, String, Integer>() {
@Override public Tuple2<String
浏览 6提问于2015-02-05得票数 3
1回答
我正在开发一个Spark流应用程序，在这个应用程序中，我需要使用Python中的两个服务器的输入流，每个服务器每秒向Spark上下文发送一个JSON消息。String> streamData2 = ssc.socketTextStream("localhost", 9995,StorageLevels.MEMORY_AND_DISK_SER);
浏览 6提问于2016-05-09得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
