我是scala和spark的新手。我想在一个函数内的spark中使用累加器来递增，因为该函数被map函数调用。我有一个RDD，它的名称是vertices，并且RDD的每一行都有一个tuple2、ID及其属性(键、值)，例如： (1,1)..
(34,1) 我想使用累加器在关键字%2等于零时递增如果它等于0，累加器将递增1，我们将有一个键等于ID的</em
浏览 25提问于2019-08-28得票数 1
回答已采纳
我想使用累加器来收集一些关于我正在操作的星火作业数据的统计数据。理想情况下，当作业计算所需的转换时，我会这样做，但是由于Spark将在不同的情况下重新计算任务，累加器将不会反映真正的指标。以下是文档对此的描述：
对于仅在操作内部执行的累加器更新，Spark保证每个任务对累加器的更新只适用一次，即重新启动<em
浏览 6提问于2015-04-07得票数 52
回答已采纳
1回答
火花蓄能器不工作
、、、
我希望使用累加器从数据中获得关闭订单的数量。但是它给了我不正确的答案，只是0。有什么问题吗？我用的是霍顿工作沙箱。代码在下面。我正在使用火花-提交。，我得到了零。spark-submit --master yarn closedCounter.pyUpDate：
rdd.foreac
浏览 1提问于2018-02-04得票数 0
回答已采纳
2回答
分类键值在HashMap中的程序开始时加载.因此，需要将每个传入的数据包与这些密钥进行比较，并相应地进行标记。如何在所有使用HashMap的火花工作者上共享我的HashMap。或者，是否有更好的<
浏览 0提问于2015-06-10得票数 8
回答已采纳
1回答
我正在使用火花蓄能器收集每条管道的统计数据。在一个典型的管道中，我会读取一个data_frame：df.write.format(delta).option("header",'true').
浏览 5提问于2021-12-08得票数 1
回答已采纳
1回答
我正在初始化一个累加器然后在map函数中，我尝试增加累加器，然后使用累加器值来设置变量。accum.add(1);
});
但是我得到了下面的错误16/03/14 09:12:58错
浏览 3提问于2016-03-14得票数 2
回答已采纳
例如，我们使用Spark执行下面的SQL，我们需要my_udf(row)返回Spark中的分区id。temporary function my_udf as 'com.my.MyUDF';
我已经知道如何让taskId在Hive中在MR engine：中执行，但是在Spark中执行它并不有效。请告诉我如何获得partitionID或taskC
浏览 3提问于2021-06-22得票数 1
回答已采纳
在地图操作中，我使用LongAccumulator作为共享计数器。但是，我似乎没有正确地使用它，因为工作节点上的计数器状态没有被更新。，当应用程序在多个工作节点中运行时，它应该工作得很好：
累加器是只通过联想和交换操作“添加”的变量，因此可以有效地并行支持。它们可以用于实现计数器(如MapReduce)或和。本机支持数值类型的累加器，程序员可以添加对新类型的支持。但是，当计数器在两个不同的
浏览 5提问于2017-05-04得票数 5
回答已采纳
1回答
我想做这样的事。在上面的代码中，the_accumulator_ojbect的位置应该是什么？写ac还行吗？同时，在函数中{}
在上面的函数中，TypeOfAccumulator的位置应该是什么？
浏览 3提问于2016-08-03得票数 4
回答已采纳
我试图使用星火累加器删除一个组的查询，它的性能很差。import org.apache.spark._在我的累加器<em
浏览 3提问于2016-02-13得票数 0
回答已采纳
1回答
我已经看过了已经提供的答案，但我仍然被卡住了。但是，当我尝试编辑批处理脚本以添加那些用引号括起来的参数时，根据我尝试添加双引号的方式和位置等，我得到了各种错误。所以我的问题可能是我不能转义引号和使用环境变量？最终
浏览 1提问于2010-02-13得票数 3
回答已采纳
我使用LongAccumulator来计算我在Cassandra中保存的记录的数量。Spark Web UI, ok too.record.data1)} }我看到代码执行正确，我将数据保存在Cassandra中，当我最终打印累加器时，值是0，但如果我在map函数中打印它，我可以看到正确</em
浏览 0提问于2018-05-10得票数 3
我想在spark中的foreachparition中执行mysql查询，并最终将所有查询结果放到一个数据帧中。看起来是这样的： val result我注意到在MysqlService中，我每次都创建db会话，这可能是不正确的。有没有更好的方法呢
浏览 52提问于2018-06-09得票数 0
1回答
在我的结构化流工作中，我正在更新updateAcrossEvents方法中的火花累加器，但是当我试图在我的StreamingListener中打印它们时，它们总是0。mapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout())( )
编辑:更详细地描述问题的更多信息累加器在“updateAcrossEvent
浏览 0提问于2020-05-14得票数 0
在Hive中定义的外部表Spark出现了一个奇怪的问题
CREATE EXTERNAL TABLE ... STORED AS PARQUET...")引用Spark中的表，就会得到正确的行计数，但是每个值都是空的。当我通过Beeline查询表时，我得到了正确的值。此外，如果我使用spark.read
浏览 1提问于2021-02-09得票数 0
我正在尝试使用spark执行一个简单的应用程序示例代码。使用spark submit执行作业。ensure that workers are registered and have sufficient memory该作业提供了正确的结果5g
spark.master spark://<master
浏览 2提问于2015-03-09得票数 0
2回答
我没有得到线性回归问题的输出。这是一个简单的单变量线性回归问题。我用过Kaggle的线性回归数据集，它没有给出期望的output.It，而是给出了权重和偏差的nan值。plt.scatter(X_train,Y_train)它正在提供产出：权重和偏倚得到了
浏览 0提问于2018-03-18得票数 1
回答已采纳
1回答
Spark 1.5.1 + Java 1.8 //accumulator.add(recoords.length); // ...在驱动程序节点上，有一个线程监视累加器值即使累加器使用延迟值设置，也应该正确地更新它，因为我在驱动程序节点线程中定期读取该值。
我是不
浏览 1提问于2016-01-26得票数 4
回答已采纳
2回答
这是我正在尝试做的事情的一个简化版本。我想在我的udf中做一些计数。这样做的一种方法是将长累加器传递给udf，并在deserializeProtobuf函数的if else循环中递增累加器。有没有更好的办法？
浏览 3提问于2021-03-23得票数 0
我给我的女儿看了两个111.111数字乘法的魔术图案。从最后一行可以看出，答案显然是错误的，应该是1234567898765432*1*，而不是1234567898765432**。有人说它可能是由整数溢出引起的。这里有两个问题：
谢谢。
浏览 3提问于2019-10-20得票数 5
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
