parallelize调用中的numSlices被后面的coalesce覆盖
在Spark中，parallelize是将一个已有的集合转化为分布式数据集（RDD）的方法。它将集合切分成多个分区，并在集群中的多个节点上并行处理这些分区。在parallelize方法中，numSlices参数用于指定切分成的分区数。
而coalesce是Spark中的一个操作，用于减少RDD的分区数。它将多个分区合并为较少的分区，以减少数据的传输和处理开销。coalesce方法可以通过传递参数shuffle来触发数据的重新分区，但默认情况下，它只是简单地合并分区而不进行数据的洗牌操作。
在给定的问答内容中，numSlices参数在parallelize调用中指定了切分成的分区数。然而，后面的coalesce操作可能会覆盖这个参数的设置，导致分区数发生变化。
如果我们想确保在coalesce操作后分区数不变，可以使用repartition方法代替coalesce方法。repartition方法会触发数据的洗牌操作，确保分区数的改变。
总结起来，parallelize调用中的numSlices参数用于指定切分成的分区数，而后面的coalesce操作可能会覆盖这个设置。如果需要保持分区数不变，可以使用repartition方法代替coalesce方法。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark服务：https://cloud.tencent.com/product/spark
相关·内容
1回答
火花中的聚结(NumPartitions)是否经过洗牌？
apache-spark
关于火花变换函数，我有一个简单的问题。
合并( numPartitions )--将RDD中的分区数量减少到numPartitions。在过滤掉大型数据集之后，对于更有效地运行操作非常有用。
val dataRDD = sc.textFile("/user/cloudera/inputfiles/records.txt")
val filterRDD = dataRDD.filter(record => record.split(0) == "USA")
val resizeRDD = filterRDD.coalesce(50)
val result
浏览 3提问于2017-02-03得票数 2
回答已采纳
5回答
java.lang.OutOfMemoryError:无法获得100个字节的内存，获得0
python、hadoop、memory、apache-spark、pyspark
我使用以下命令在本地模式下使用Spark2.0调用Pyspark：
pyspark --executor-memory 4g --driver-memory 4g
输入数据正在从tsv文件中读取，并且有580 Kx28列。我正在对dataframe做一些操作，然后我尝试将它导出到tsv文件中，我得到了这个错误。
df.coalesce(1).write.save("sample.tsv",format = "csv",header = 'true', delimiter = '\t')
任何关于如何消除这个错误的提示。我可以很容易
浏览 7提问于2016-08-15得票数 24
回答已采纳
20回答
火花-重新分区()与coalesce()
apache-spark、distributed-computing、rdd
根据学习火花
请记住，重新分区您的数据是一个相当昂贵的操作。Spark还有一个名为repartition()的优化版本coalesce()，它允许避免数据移动，但前提是要减少RDD分区的数量。
我发现的一个不同之处是，使用repartition()，分区数量可以增加/减少，而使用coalesce()，分区数只能减少。
如果分区分布在多台机器上，并且运行coalesce()，那么如何避免数据移动？
浏览 6提问于2015-07-24得票数 391
回答已采纳
1回答
火花-在RDD.map中将多个核分配给一个任务
apache-spark、parallel-processing、pyspark、spark-dataframe、rdd
我刚开始使用RDD.map来并行运行火花框架中的一个名为function的方法(在一个独立的火花集群中总共有72个核--一个带有100 each的驱动程序和3个工作人员，每个驱动程序都有24核和100 G RAM)。
我的目标是运行200次函数，平均运行结果。这个函数的输出是一个大小为12的numpy.array，由num_of_samples (就内存而言是一个巨大的变量)。
我的第一次尝试是创建一个大小为200的RDD，然后使用RDD.map并在最后减少：
sum_data = sc.parallelize(range(0,200)).map(function).reduce(lambda
浏览 1提问于2017-05-16得票数 2
1回答
为什么Spark总是将相同数量的文件写入HDFS？
apache-spark-sql、hdfs、spark-streaming
我有一个用Scala编写的Spark流媒体应用程序，在CDH中运行。应用程序从Kafka读取数据并将数据写入HDFS。在将数据写入HDFS之前，我执行了partitionBy，因此数据是分区写入的。每个分区在写入时都会得到3个文件。我还使用coalesce来控制数据的分区数量。我的预期是，由coalesce命令设置的分区数量将设置HDFS中输出目录中的文件数量，然而，尽管由coalesce命令设置了分区数量，文件数量始终是3。我尝试使用3个执行器和6个执行器运行，但每个分区中的文件数量仍然是3个。
这就是我将数据写入HDFS的方式：
//Some code
val ssc = new Stre
浏览 0提问于2018-09-18得票数 1
1回答
重新划分(1)是否一直保持秩序？
scala、apache-spark、rdd
我需要压缩两个可能有或可能没有相同分区的rdd，因此寻找重新分区的方法。我需要保持秩序，同时拉链，我知道重新划分洗牌一般。但下面的代码显示，repartiton(1)并不是在对rdd进行洗牌。是只有这一次，还是我们每次都能保证？
重分区(1)是否与.collect相似，因为它们都将rdd带到单个节点？
scala> var k = sc.parallelize((1 to 100),4)
k: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:27
sca
浏览 0提问于2017-05-19得票数 0
2回答
如何控制在spark中设置大分区时生成的文件数？
apache-spark、pyspark、apache-spark-sql
由于输入数据的数量很大，我设置了spark (spark.sql.shuffle.partitions=1000)的大洗牌分区。但是，输出文件很小(~1GB)，但是它会创建大量的小文件(3000个文件，每个文件小于1Mb)。如何将这些小文件组合成一个大文件？
另一个问题是，为什么输出文件的数量是洗牌分区数量的3倍？
浏览 24提问于2021-12-14得票数 1
1回答
Apache Spark join操作的弱伸缩性差
performance、scala、apache-spark、distributed-computing
我在Apache Spark上运行"join“操作，发现没有弱可伸缩性。如果有人能解释这一点，我将不胜感激。
我创建了两个数据帧("a"，"b")和("a"，"c")，并通过第一列连接这两个数据帧。我为“一对一”连接生成数据帧值。此外，我使用相同的分割器来避免混洗。
数据帧中的行数- 1024 * 1024 * 16 * cores_total (cores_total -在其上启动程序的核心总数)。列"a“由随机Int值组成，"b”列的所有值等于1，"c“列的所有值等于2。
理论上，随着数据大小和
浏览 0提问于2017-10-04得票数 3
2回答
在dataframe中使用dropDuplicates会导致分区号的更改。
apache-spark、pyspark、apache-spark-sql、partitioning
我有一个很大的dataframe，我用800个分区创建了它。
df.rdd.getNumPartitions()
800
当我在dataframe上使用dropDuplicates时，它将分区更改为默认的200。
df = df.dropDuplicates()
df.rdd.getNumPartitions()
200
这种行为给我带来了问题，因为它会导致我失去记忆。
你对解决这个问题有什么建议吗？我试着将spark.sql.shuffle.partition设置为800，但它不起作用。谢谢
浏览 5提问于2016-05-26得票数 4
回答已采纳
3回答
在将RDD数据保存到文件之前，是否需要使用合并？
scala、apache-spark
假设我有一个有100条记录的RDD，我用10条对它进行了分区，所以现在每个分区都有10条记录--我只是将其转换为rdd到键值对rdd，然后将它保存到一个文件中--现在我的输出数据被分割成10个分区，这对我来说是可以的，但在将输出数据保存到文件之前，最好先使用coalesce函数吗？例如，rdd.coalesce(1)，这只给出一个文件作为输出，它不是对节点内部的数据进行洗牌吗？想知道哪里应该使用聚结。
谢谢
浏览 4提问于2015-10-29得票数 1
回答已采纳
1回答
控制星火中任务的分布
java、apache-spark
我正在星火库集群上分发一些下载任务。输入来自一个源，它不能总是与Spark的常规方法(如parallelize或textFile等)并行。相反，我有一个服务提供给我一堆我使用parallelize分发的下载任务(URL +封装逻辑来读取和解密它)。
当有几千个任务时，Spark将任务平均分配给所有的奴隶，从而实现最大程度的并行性。然而，当有几百个任务时，Spark认为数据集很小，可以在几个从节点上进行计算，以减少通信时间，增加数据局部性。但在我的例子中，这是错误的，每个任务都可以生成数以千计的JSON记录，我希望下载的机器和集群中的机器一样多。
我现在有两个想法：
使用repartitio
浏览 1提问于2015-06-27得票数 0
回答已采纳
1回答
不带列参数的df.repartition在什么上分区？
python、apache-spark、pyspark、pyspark-sql
在PySpark中，repartition模块有一个可选的columns参数，它当然会按该键对数据帧进行重新分区。
我的问题是- Spark如何在没有键的情况下重新分区？我无法更深入地挖掘源代码，找出Spark本身到底在哪里。
def repartition(self, numPartitions, *cols):
"""
Returns a new :class:`DataFrame` partitioned by the given partitioning expressions. The
resulting DataFrame is ha
浏览 6提问于2018-11-29得票数 10
1回答
超长队列的spark任务是如何处理的？
apache-spark、pyspark、hadoop-yarn
我使用pyspark处理多个日志文件，其中一条记录被分成多行格式，所以我选择wholeTextFiles来读取数据，然后过滤出我想要的内容。每个文件大小约800M，共有4096个文件。然而，spark作业在处理一些任务后崩溃，以下是我的配置和代码。
配置：
--num-executors 100 --executor-cores 1 --executor-memory 30G
核心代码：
file_rdd= sc.wholeTextFiles(inputDir, 2500)
print file_rdd.getNumPartitions()
out_rdd = file_rdd.flatMap
浏览 20提问于2019-10-28得票数 0
1回答
如何在Spark Rdd中转换Seq
scala、apache-spark、playframework
我正在使用Spark Scala和Play框架，我有一个这样的seq
//a sequence of Book objects
val books:[Seq[Book]]
我用json文件中的format方法填充：
implicit val bookFormat: Format[Libri] = {
((JsPath \ "City").format[String] and
(JsPath \ "GEN").format[Int] and
(JsPath \ "SER").format[Int]
) (Lib
浏览 0提问于2016-10-19得票数 3
1回答
如何在蔚蓝数据库中平均分割300万条记录
pyspark、azure-databricks、azure-eventhub
我有一个连接到oracle数据库并读取有300万条记录的表的。我需要把这张数据写成蓝色的。
下面是样例吡火花数据框写入到甚至是。
df.select("body") \
.write\
.format("eventhubs") \
.options(**ehconf) \
.save()
如何将我的电火花数据分割成10部分(300 k记录/数据)？
这样，我就可以将这10条电火花数据文件中的每一条都发送给eventhub。
浏览 4提问于2022-04-08得票数 0
1回答
如何提高火花性能？
java、apache-spark、cassandra、hdfs、spark-cassandra-connector
我有处理大型数据集的Java程序。数据集存储在hdfs (csv)中。
这个程序运行得很好，但速度很慢。
该方案是做什么的：
String[]filter字符串数组加载csv文件单独行映射到MyObject 保存MyObject到Cassandra<code>G 211</code>
我的主要方法是：
public static void main(String[] args) {
// configure spark
SparkConf sparkConf = new SparkConf().setAppName("Write
浏览 6提问于2020-05-22得票数 3
回答已采纳
2回答
最简单程序的大任务大小
scala、apache-spark、apache-spark-sql
我试着用星火运行最简单的程序
import org.apache.spark.{SparkContext, SparkConf}
object LargeTaskTest {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("DataTest").setMaster("local[*]")
val sc = new SparkContext(conf)
val dat = (1 to 10000000).toList
val dat
浏览 3提问于2015-11-24得票数 5
回答已采纳
2回答
重分区不影响任务数。
python、apache-spark、pyspark
如何增加任务的数量，以减少每个任务所需的内存量？
以下非常简单的示例失败：
df = (
spark
.read
.format('delta')
.load(input_path)
)
df = df.orderBy("contigName", "start", "end")
# write ordered dataset back to disk:
(
df
.write
.format("delta")
.save(output_path)
)
浏览 5提问于2019-12-20得票数 0
2回答
Spark 2.0读取csv分区数(PySpark)
csv、apache-spark、pyspark
我正在尝试使用Spark 2.0中的新东西将一些代码从Spark 1.6移植到Spark 2.0。首先，我想使用Spark 2.0的csv阅读器。顺便说一句，我用的是pyspark。
使用“旧的”textFile函数，我可以设置分区的最小数量。例如：
file= sc.textFile('/home/xpto/text.csv', minPartitions=10)
header = file.first() #extract header
data = file.filter(lambda x:x !=header) #csv without header
...
现在，使用
浏览 0提问于2016-07-01得票数 5
回答已采纳
2回答
重新划分(1)与合并(1)之间的差异
apache-spark、apache-spark-sql
在我们的项目中，我们使用repartition(1)将数据写入表，我很想知道为什么这里不能使用coalesce(1)，因为与coalesce相比，repartition是一种昂贵的操作。
我知道repartition在分区间平均分配数据，但是当输出文件是单个部件文件时，为什么不能使用coalesce(1)呢？
浏览 6提问于2021-09-12得票数 4
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
即时通信 IM
ICP备案
对象存储
实时音视频活动推荐
腾讯云开发者
