1回答
当我执行以下命令时：rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[10] at partitionBy at <consolescala> rdd.parti
浏览 1提问于2015-12-28得票数 20
回答已采纳
在阅读完它们之后，我需要执行一系列的转换。其中之一就是引爆一根柱子。在转换之后，我得到了下面描述的洗牌溢出。我明白为什么会这样。爆炸是基于一个广播变量查找，它提供了一个非常扭曲的结果集。我尝试在explode函数之前重新分区，方法是调优spark.sql.shuffle.partitions配置参数，以确保洗牌分区大小相等，但这没有帮助。
浏览 1提问于2020-04-01得票数 0
回答已采纳
我希望尽可能避免按键对数据集进行重新分区，并知道给定键的所有记录是否已经在同一分区中。
Spark中有没有一个内置的函数可以给我答案？
浏览 2提问于2016-12-29得票数 7
学习自定义Spark RDD分区，编写了一些逻辑，但不编译。在Spark 2.4.3中，启动spark shell：val transactions= sc.parallelize(transactions).partitionBy(new CountryPartitioner(2))error: value
浏览 20提问于2019-08-19得票数 0
回答已采纳
一个RDD在5-1000万个条目之间，另一个RDD在5亿到7.5亿个条目之间。在某种程度上，我必须使用公共密钥连接这两个rdd。, y); } // 600-million当spark决定进行此连接时，它决定执行一个ShuffledHashJoin。更新7/14
我的性能问题似乎根植于重新分区。通常，从HDFS读取的RDD将按块进行分
浏览 4提问于2015-07-13得票数 11
回答已采纳
我有一个用Scala编写的Spark流媒体应用程序，在CDH中运行。应用程序从Kafka读取数据并将数据写入HDFS。在将数据写入HDFS之前，我执行了partitionBy，因此数据是分区写入的。每个分区在写入时都会得到3个文件。我还使用coalesce来控制数据的分区数量。我的预期是，由coalesce命令设置的分区数量将设置HDFS中输出目录中</e
浏览 0提问于2018-09-18得票数 1
1回答
我想澄清一下DAG的行为，以及具体如何处理以下工作：.partitionBy(new HashPartitioner(4))
.partitionByParallelCollectionRDD[2] at parallelize at
浏览 3提问于2017-03-23得票数 2
回答已采纳
根据答案，密钥的所有数据都驻留在一个分区中。但是，这个来自星火邮件组的回答却有所不同。
请有人确认一个分区的所有数据是否在一个任务分区中结束。
浏览 4提问于2017-03-13得票数 1
2回答
我正在使用Spark的Python并运行Spark0.8。RDD中的片和分区之间有什么区别吗？当我创建RDD时，我将它作为一个参数传递给它，这将导致它将RDD存储为100片，并在执行计算时创建100个任务。我想知道，通过使系统能够更有效地处理数据，对数据进行分区是否会提高切片以外的性能(即，对分区执行操作与仅对切片的RDD<e
浏览 3提问于2014-06-17得票数 12
回答已采纳
众所周知，Spark中的分区对任何“宽”操作都有巨大的性能影响，所以它通常是在操作中定制的。我正在使用以下代码进行实验： sc.parallelize(1 to 50).keyBy(_ % 10)val rdd2 =
val cogrouped = <em
浏览 48提问于2015-05-01得票数 26
回答已采纳
我有一个应用程序，其中有大约10个平面文件，每个文件的价值都超过200MM+记录。业务逻辑涉及顺序地将它们全部连接起来。我的环境:1主-3从(为了测试，我为每个节点分配了1 1GB内存) RDD1 = sc.textFile(file1).mapToPair(..)RDD2 = sc.textFile(file2).mapToPair(..)join = RDD1.join(RDD2).map(peopleObject)
有没有关于调整的建议，比如重新分区<
浏览 2提问于2016-05-30得票数 5
当我使用reduceByKey或aggregateByKey时，我会遇到分区问题。例如，reduceBykey(_+_).map(code)因此，作为解决方案，我使用了repartition方法。这对于分区分发是有好处的，但是repartition也很昂贵。
有没有办法明智地解决分区问题？
浏览 1提问于2017-03-26得票数 5
我是Spark的新手。考虑以下代码： val rdd = sc .partitionBy(new HashPartitioner(sc.defaultParallelism))rdd.count() 从文件中读取的每个元组是否直接发送到散列分区程序指定的分区？或者，整个文件首先被读取到内存中，而不考虑分区</
浏览 20提问于2020-04-25得票数 0
我正在寻找一种方法来写入和恢复分区数据集。对于这个问题，我可以接受两个分区的RDD和Dataset[Row] / Dataframe这样做的目的是为了在恢复数据时避免混洗例如：
spark.range(
浏览 0提问于2016-10-06得票数 2
我有一个主程序和两个从程序，每个主程序和两个从程序都运行在32 GB的RAM上，并且我正在读取一个包含大约1800万条记录的csv文件(第一行是列的标题)。./spark-submit --master yarn --deploy-mode client --executor-memory 10g <path/to/.py file>rdd= sc.textFile("<path&#x
浏览 0提问于2015-10-12得票数 2
我有一个包含30条记录的RDD (键/值对:键是时间戳，值是JPEG字节数组)
我管理着30个执行者。我想将这个RDD重新划分为30个分区，这样每个分区就可以得到一条记录，并被分配给一个executor。当我使用rdd.repartition(30)时，它将我的rdd重新划分为30个分区，但一些分区获得2条记录，一些分区获得1条记录，而另一些则没有任何记录。有没有办法在Spark中
浏览 1提问于2015-11-18得票数 5
1回答
我正在使用partitionBy()在Spark中进行散列分区。但是该应用程序接口在RDD class.But中不可用，它显示无法解析在单机上运行的partitionBy() .I am，并且Spark核心版本为1.2。import org.apache.spark.SparkContext(sparkConf);
val testrdd = sparkContext.parall
浏览 0提问于2016-04-05得票数 0
我有两个我想加入的RDD，它们看起来像这样：val rdd2:RDD[((T,W), V)]val rdd_joined:RDD[((T,W), (U,V))]val m = rd
浏览 0提问于2013-07-13得票数 49
回答已采纳
3回答
从读取查找方法的定义：lookup(key: K): Seq[V]
Return the list of values in the RDD for key key.This operation is done efficiently if the RDD has a known partitioner by only searching the partition如何确保RDD有一个已知的分区程序？我知道RDD是跨集群中的节点进行分区的，但
浏览 1提问于2015-05-07得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
