如何在Spark中将每个列重新分区为固定数量的分区？
在Spark中，可以使用repartition方法将每个列重新分区为固定数量的分区。repartition方法是一种广义的分区操作，它可以根据指定的分区数重新分配数据，并且可以在数据重分区的同时进行数据重组。
具体步骤如下：
- 首先，创建一个SparkSession对象，用于与Spark集群进行交互。
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Repartition Example").getOrCreate()- 然后，读取数据源文件，可以使用SparkSession的read方法来加载数据。
df = spark.read.format("csv").option("header", "true").load("data.csv")- 接下来，使用repartition方法将每个列重新分区为固定数量的分区。可以通过指定列名或列的索引来进行分区。
num_partitions = 4
df_repartitioned = df.repartition(num_partitions, "column_name")- 最后，可以将重新分区后的数据保存到文件或进行进一步的处理。
df_repartitioned.write.format("csv").mode("overwrite").save("repartitioned_data.csv")在上述代码中，需要将"column_name"替换为要重新分区的列名，将"data.csv"替换为实际的数据源文件路径，将"repartitioned_data.csv"替换为保存重新分区后数据的文件路径。
需要注意的是，repartition方法会进行数据的全量洗牌操作，因此在处理大规模数据时可能会产生较高的性能开销。如果只需要对特定列进行分区，可以使用repartitionByRange方法，该方法可以根据指定的列值范围进行分区，避免全量洗牌操作。
推荐的腾讯云相关产品：腾讯云EMR（Elastic MapReduce），是一种大数据处理和分析的云服务，可以方便地在云端进行Spark集群的创建和管理。您可以通过以下链接了解更多关于腾讯云EMR的信息：腾讯云EMR产品介绍
请注意，以上答案仅供参考，具体实现方式可能因Spark版本和具体需求而有所不同。
相关·内容
1回答
我需要将数据集数据划分为6列: region/year/month/day/id/quadkey，其中在顶层我只有二进制区域状态，而在最底层实际上是它进入许多分区的位置。假设我们有两个区域/通常是一年/通常是一个月/3-4天/100-150个I/ 50-200个四键，当我执行这个操作时，我得到了非常不平衡的随机操作，有时执行器会因为超出内存限制而失败。此外，我从History UI中注意到，hat阶段的一些任务非常大(约15 1Gb)，而其他任务则小得多(约1 1Gb)。
我试着去玩
sqlContext.setConf("spark.sql.shuffle.partitions
浏览 0提问于2019-04-11得票数 1
1回答
我是一名Apache Spark/Redis用户，最近我在一个项目中尝试了spark-redis。该程序正在生成大约有300万行的PySpark数据帧，我使用以下命令将其写入到Redis数据库中 df.write \
.format("org.apache.spark.sql.redis") \
.option("table", "person") \
.option("key.column", "name") \
.save() 就像GitHub project dataframe page上建
浏览 86提问于2019-01-24得票数 0
我有一个spark应用程序，它将处理后的输出转储到S3存储桶中。我的输出数据帧在pyspark中的分区数是300。我有列"id“(大约250个唯一值)，我用它来用partitionBy写文件。因此，我得到了大约75K个文件，总和为11 to。
在我的spark应用程序中，将此数据写入S3的所有阶段都已完成，但从_temp文件夹中移动文件仍在进行中。这是非常缓慢的，按照目前的速度，这将需要一天以上的时间。这里的根本原因是什么?如何优化它？
浏览 1提问于2020-06-05得票数 0
1回答
星火拼花数据帧分区数目
、、、
我有一个HDFS文件夹与两个250 an的地板文件。hadoop块大小设置为128 to。具有以下代码：
JavaSparkContext sparkContext = new JavaSparkContext();
SQLContext sqlContext = new SQLContext(sparkContext);
DataFrame dataFrame = sqlContext.read().parquet("hdfs:////user/test/parquet-folder");
LOGGER.info("Nr. of rd
浏览 3提问于2017-06-28得票数 4
1回答
星火中的分区与数据湖中的分区
、、、、
当将文件写入Data时，特别是通过Databricks，我们可以选择指定分区列。这将根据数据集中该列中可用的值将数据保存在单独的文件夹(分区)中。
同时，当我们讨论星火优化时，我们谈到了数据的分区。
这两者有什么区别呢？他们之间有关系吗？
据我所理解，如果我们只想读取数据的某些部分(当然是基于分区列)，将数据保存在分区中的分布式文件系统中将有所帮助。例如，如果我们按颜色进行分区，并且只对“红色”记录感兴趣，我们只能在该分区中读取，而忽略其余的。这会在读取数据时产生一定程度的优化。
然后，为了让Spark执行并行处理，这个“red”分区(来自文件系统)将根据集群中可用的核数划分为分区(Spark
浏览 2提问于2022-05-10得票数 0
回答已采纳
1回答
我知道我可以repartition一个RDD来增加它的分区，并使用合并来减少它的分区。我有两个问题，在阅读了不同的资源后，我无法完全理解。
在生成RDD时，Spark将使用合理的默认设置(每个块一个分区，最初版本为64 in，现在为128 in)。但我也读到，建议使用2到3倍的核心运行作业。所以问题来了：
对于给定的文件，我应该使用多少分区？例如，假设我有一个10 3gb的.parquet文件，3个带有2个内核的执行器和每个3GB的内存。我应该重新划分吗？我应该使用多少分区？有什么更好的方式来做出这样的选择？
所有数据类型(如.txt、.parquet等)默认情况下，如果不提供分区，则
浏览 4提问于2017-08-18得票数 9
回答已采纳
1回答
我正在尝试使用以下配置在纱线集群中执行星火作业。
/usr/bin/spark-submit
--class com.example.DriverClass
--master yarn-cluster
app.jar
hdfs:///user/spark/file1.parquet
hdfs:///user/spark/file2.parquet
hdfs:///user/spark/output
20151217052915
--num-executors 20
--executor-memory 12288M
--executor-cores 5
--driver-m
浏览 2提问于2015-12-19得票数 2
1回答
我们正在试验在我们的项目中运行Spark，没有Hadoop，也没有像HDFS这样的分布式存储。Spark安装在具有10核和16 on的单个节点上，并且该节点不属于任何群集。假设Spark driver占用2个内核，其余的在执行时被执行器(每个2个)消耗。 如果我们处理存储在Spark本地磁盘中的一个大CSV文件(大小为1 GB)作为RDD，并将其重新分区到4个不同的分区，执行器是否会并行处理每个分区？如果我们不将RDD重新划分为4个不同的分区，执行器会做什么？如果不使用HDFS，我们会失去分布式计算和并行性的能力吗？
浏览 10提问于2019-02-01得票数 0
回答已采纳
我有一段创建DataFrame并将其持久化到S3的代码。创建一个由math.Random填充的1000行100列的DataFrame。我在一个有4个r3.8xlarge工作节点的集群上运行，并配置了足够的内存。我已经尝试了最大数量的执行器，并且每个节点有一个执行器。
// create some random data for performance and scalability testing
val df = sqlContext.range(0,1000).map(x =>
Row.fromSeq((1 to 100).map(y => math
浏览 2提问于2015-09-09得票数 1
1回答
假设我正在从一个S3文件夹读取100个文件。每个文件的大小为10 MB。当我执行df = spark.read.parquet(s3 path)时，文件(或者更确切地说是分区)是如何跨任务分发的？例如，在本例中，df将有100个分区，如果spark有10个任务用于将该文件夹的内容读取到数据框架中，那么分区是如何分配给这10个任务的？它是以循环的方式进行的，还是每个任务在基于范围的分发中获得的所有分区的比例相等，或者其他什么？任何指向相关资源的指针也将非常有用。谢谢。
浏览 3提问于2019-12-09得票数 2
回答已采纳
2回答
我是Spark和Cassandra的新手
我们在Cassandra之上使用Spark来读取数据，因为我们需要使用非主键列读取数据。
一种观察是，spark作业的任务数量增加了w.r.t数据增长。正因为如此，我们在获取数据时面临着大量的延迟。
spark作业任务数增加的原因是什么？
应该考虑什么来提高Cassandra在Spark中的性能？
请推荐我。
谢谢,
马利卡金
浏览 1提问于2015-09-08得票数 2
1回答
使用，使用Cloudera解决方案，我使用的是3个执行器，但是当我启动应用程序时，使用的执行器只有一个。如何使用多个执行器？
让我给你更多信息。这是我的参数：
命令启动：
spark2-submit --master yarn \
--deploy-mode cluster \
--conf spark.ui.port=4042 \
--conf spark.eventLog.enabled=false \
--conf spark.dynamicAllocation.enabled=false \
--conf spark.streaming.backpressure.enabled=tru
浏览 0提问于2018-04-27得票数 2
回答已采纳
在对数据的某一列执行聚合操作之前，我正在尝试对数据进行预分区。我有3个工作节点，我希望每个分区在我要分区的列中有不重叠的值。我不希望出现两个分区在列中可能具有相同值的情况。
如果我有以下数据
ss_item_sk | ss_quantity
1 | 10.0
1 | 4.0
2 | 3.0
3 | 5.0
4 | 8.0
5 | 13.0
5 | 10.0
然后，以下分区是令人满意的：
分区1
ss_item_sk | ss_quantity
1
浏览 1提问于2019-01-06得票数 0
回答已采纳
我必须首先按“客户组”进行分区，但我也希望确保每个"customer_group“都有一个csv文件。这是因为推断所需的是时间序列数据，不能跨多个文件传播。
我试过：datasink2 = spark_df1.write.format("csv").partitionBy('customer_group').option("compression","gzip").save(destination_path+'/traintestcsvzippartitionocalesce')，但它在customer_g
浏览 6提问于2022-05-14得票数 0
回答已采纳
我正在尝试在Spark Structured和Hazelcast Jet之间映射一些概念，并理解其他主题。
Q1 -在Spark中，每个Kafka分区将成为spark中的一个分区，然后它们将由单个任务并行处理。我想我在某处读到过Hazelcast Jet将合并来自kafka的所有消息，而不考虑group.id和主题分区，对吗？
Q2 -我们如何在Jet程序中增加“消费者”的数量，以增加来自kafka的吞吐量消耗？在Spark中，我想我们只需要增加主题分区的数量，以便为新分区分配新的spark任务。
Q3 -如果上面的Q1是真的，有可能避免合并和分发要并行处理的kafka分区吗？一旦消息已经在k
浏览 0提问于2018-01-23得票数 2
假设我有一个大小为100MB的输入文件。它包含CSV格式的大量点(经纬度对)。我应该做什么，以便在Apache Spark中将输入文件拆分为10个10MB文件，或者如何自定义拆分。
注意:我希望处理每个映射器中的点的子集。
浏览 2提问于2014-12-23得票数 4
回答已采纳
3回答
问题：--我希望使用以下方法从S3中将数据导入Spark：
data = sqlContext.read.json("s3n://.....")
是否有一种方法可以设置Spark用于加载和处理数据的节点数？这是我如何处理数据的一个例子：
data.registerTempTable("table")
SqlData = sqlContext.sql("SELECT * FROM table")
Context：数据不太大，加载到Spark需要很长时间，也需要很长时间来查询。我认为Spark将数据划分为太多的节点。我希望能够手动设置。我知道在处理R
浏览 0提问于2016-01-04得票数 17
我在Spark中有一个Dataframe，它注册为一个名为A的表，并且有1 billion记录和10列。第一列(ID)是主键。
还有另一个Dataframe，它注册为一个名为B的表，具有10,000记录和10列(与表A相同的列，第一列(ID)是主键)。
表B中的记录是“更新记录”。因此，我需要用表B中的记录更新表A中的所有10,000条记录。
我首先尝试了这个SQL查询：
select * from A where ID not in (select ID from B)，然后将其与表B合并。
方法还可以，但是第一个查询(select * from A where ID not in (sel
浏览 0提问于2020-01-22得票数 0
我使用SparkSQL处理数据，并且我想将我的数据写入一个子文件。
...
step12.show()
step12.repartition(10).coalesce(1).write.json('wasb://liu@cliubo.blob.core.windows.net/test_data_4')
step12是我的数据帧，但是我收到一个错误，告诉我java.lang.OutOfMemoryError: Unable to acquire 65536 bytes of memory, got 0它没有意义，因为我可以显示这个数据帧。我使用Microsoft Azure中的
浏览 0提问于2017-04-17得票数 0
1回答
我对Spark SQL有个问题。我从csv文件中读取了一些数据。接下来，我执行groupBy和join操作，完成的任务是将连接的数据写入文件。我的问题是时间差距(我在下面的日志中用空间显示了这一点)。
18/08/07 23:39:40 INFO spark.ContextCleaner: Cleaned accumulator 1069
18/08/07 23:39:40 INFO spark.ContextCleaner: Cleaned accumulator 1003
18/08/07 23:39:40 INFO spark.ContextCleaner: Cleaned accumu
浏览 12提问于2018-08-10得票数 1
2回答
我有大量的最终结果数据，这些数据在我感兴趣的列中分布得并不均匀。当我通过分区直接写入时，每个分区具有与spark.sql.shuffle.partitions相同数量的文件。这会导致拥挤分区中的每个文件都非常大(以GB为单位)，但在其他一些分区中，文件大小实际上很小(甚至以KB为单位)。有没有办法改变每个分区的文件数？
示例：
+----------------------------+----------+
| number of rows in category | category |
+----------------------------+----------+
| 5000000
浏览 2提问于2018-12-05得票数 0
我正在尝试修复在我的spark设置中看到的一个外存问题，在这一点上，我无法得出一个具体的分析，为什么我会看到这个。在将数据帧写到parquet或kafka时，我总是会遇到这个问题。我的数据帧有5000行。它的模式是
root
|-- A: string (nullable = true)
|-- B: string (nullable = true)
|-- C: string (nullable = true)
|-- D: array (nullable = true)
| |-- element: string (contains
浏览 0提问于2017-09-14得票数 0
因此，我理解在通常情况下，在以下情况下应该使用coalesce()：
由于filter或其他可能导致减少原始数据集(RDD、DF)的操作，分区数量减少。coalesce()对于过滤大型数据集后更有效地运行操作非常有用。
我还了解到，它比repartition便宜，因为它只在必要时通过移动数据来减少洗牌。我的问题是如何定义coalesce接受的参数(idealPartionionNo)。我正在做一个从另一个工程师那里传递给我的项目，他用下面的计算来计算这个参数的值。
// DEFINE OPTIMAL PARTITION NUMBER
implicit val NO_OF_EXECUTO
浏览 1提问于2016-11-29得票数 20
回答已采纳
在纱线上运行spark作业后，作业在大约9小时后失败，并显示
org.apache.spark.SparkException: Task failed while writing rows
at org.apache.spark.sql.execution.datasources.DynamicPartitionWriterContainer.writeRows(WriterContainer.scala:446)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand$$anonfun$r
浏览 1提问于2017-06-13得票数 2
我可能对join / groupBy-agg有一个天真的问题。在RDD的日子里，每当我想要执行.groupBy-agg时，我都会说reduceByKey (对于PairRDDFunctions)和一个可选的分区策略(使用的是分区的数量或分区工具) b. join (对于PairRDDFunctions)及其变体，我曾经有一种方法可以提供许多分区
在DataFrame中，如何指定此操作过程中的分区数量？我可以在事后使用repartition()。但这将是工作的另一个阶段。
在join期间增加分区/任务数量的一种变通方法是在spark-submit期间设置“spark.sql.shuffle.pa
浏览 0提问于2017-07-12得票数 1
我在我的Spark代码中使用HiveContext运行hive查询。无论我运行哪个查询，以及它有多少数据，它总是生成31个分区。有人知道原因吗？是否有预定义/可配置的设置？我基本上需要更多的分区。
我使用以下代码片段来执行配置单元查询：
var pairedRDD = hqlContext.sql(hql).rdd.map(...)
我使用的是Spark 1.3.1
谢谢你，妮汀
浏览 3提问于2016-04-28得票数 1
df = spark.read.format().load(path)
df.write.mode(WriteMode.Overwrite).format().partitionBy("ColumnX").save(path)
注意: ColumnX有12个不同的值
df_partitioned = spark.read.format().load(path)
print(df_partitioned.rdd.getNumPartitions())
output: 80
为什么我得到80而不是12
浏览 19提问于2020-08-03得票数 1
回答已采纳
各位，
当我试图在每行一个键标识符上加入2个大数据(100 to +每个)时，我遇到了这个问题。
我在EMR上使用Spark1.6，下面是我正在做的事情：
val df1 = sqlContext.read.json("hdfs:///df1/")
val df2 = sqlContext.read.json("hdfs:///df2/")
// clean up and filter steps later
df1.registerTempTable("df1")
df2.registerTempTable("df2")
浏览 1提问于2016-11-25得票数 7
我正在处理一个奇怪的情况，我有小表和大表要使用spark处理，而且它必须是一个spark工作。
为了实现最佳性能目标，我需要设置一个名为
spark.sql.shuffle.partitions = 12 for small tables and
spark.sql.shuffle.partitions = 500 for bigger tables
我想知道如何在spark中动态更改这些属性？我可以有多个配置文件并在程序中调用它吗？
浏览 4提问于2018-03-07得票数 1
我有几个Spark作业，每天处理数千个文件。文件大小可能从MB到GB不等。完成工作后，我通常使用以下代码进行保存
finalJavaRDD.saveAsParquetFile("/path/in/hdfs"); OR
dataFrame.write.format("orc").save("/path/in/hdfs") //storing as ORC file as of Spark 1.4
Spark job在最终输出目录中创建了大量的小零件文件。据我所知，Spark为每个分区/任务创建零件文件，如果我错了，请纠正我。我们如何控制Spark创
浏览 4提问于2015-07-06得票数 2
回答已采纳
1回答
下面的并行算法是使用Spark和Java实现的。但我想知道如何指定从机器中可以使用的内核数量来查看性能。起初，我认为我可以在setMaster("local[2]")中指定它可以使用的核心数量。但我觉得不管用。
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.commons.lang.time.StopWatch;
import java
浏览 0提问于2021-01-08得票数 0
回答已采纳
为下面的字数统计程序创建的分区是10，但根据我的理解，如果我们在创建sparksession对象时设置了master("local2")，这意味着它将在本地运行2个内核，即2个分区
谁能帮我解释一下为什么我的spark代码创建了10个分区而不是2个。
代码：
SparkSession spark = SparkSession.builder().appName("JavaWordCount").master("local[2]").getOrCreate();
JavaRDD<String> lines = spar
浏览 0提问于2018-12-23得票数 0
我正在尝试学习使用Dataframes编程。在下面的代码中，我尝试将两个CSV连接到一个列上，然后将其保存为一个组合CSV。在SCALA中运行这段代码的结果是，我看到了将近200个小部件文件。你能帮我理解一下这里出了什么问题吗-
import org.apache.spark.SparkContext
object JoinData {
def main(args: Array[String]) {
val sc = new SparkContext(args(0), "Csv Joining example")
val sqlContex
浏览 1提问于2016-04-19得票数 2
在解决了这个问题之后：How to limit FPGrowth itemesets to just 2 or 3，我正在尝试将使用pyspark的fpgrowth的关联规则输出导出到python中的.csv文件。在运行了近8-10小时后，它给出了一个错误。我的机器有足够的空间和内存。 Association Rule output is like this:
Antecedent Consequent Lift
['A','B'] ['C']
浏览 19提问于2019-07-02得票数 2
我正在尝试对评论做感伤的分析。程序在Spark上成功运行，但我面临的问题是在70个分区中，68个分区的时间比最后2个分区的时间缩短了约20%。我已经检查了我的数据在所有分区上是否均匀分布，甚至还检查了不同的样本数据。
另外，我使用persist(StorageLevel.MEMORY_AND_DISK_SER)为所有数据帧运行了代码，并在不再需要这些数据帧时立即取消它们的持久化。
我也尝试了增加和减少分区的数量，但对于最后两个任务，它仍然需要大量的时间。下面是我正在使用的当前配置
--master yarn \
--deploy-mode client \
--num-executors 15
浏览 29提问于2016-09-26得票数 1
我已经设置了一个包含4个worker(每个有4个核心)和1个master的spark独立集群。每台都有Windows10操作系统。我向我们的spark独立集群提交了spark的ML示例: multilayer_perceptron_classification.py。但它是将所有任务都交给一个执行者在一个worker上执行。enter image description here 火花代码是(代码使用multilayer_perceptron_classification.py MLlib)： spark = SparkSession\
.builder.appName("m
浏览 33提问于2019-02-26得票数 0
回答已采纳
我正在从spark的cassandra表中读取一张表。我在cassandra中有很大的分区，当cassandra的分区大小超过64MB时，在这种情况下，cassandra分区将等于spark分区。由于分区太大，我在spark中遇到了内存问题。
我的问题是，如果我在从cassandra读取数据后开始重新分区，spark分区的数量会发生变化吗？它不会导致火花记忆问题吗？
我的假设是，spark首先会从cassandra读取数据，因此在这个阶段，cassandra的大分区不会因为重新分区而拆分。重新分区将处理从cassandra加载的底层数据。
我只是想知道，当从spark读取数据时，重新分区是否会
浏览 48提问于2021-06-07得票数 1
我正在尝试读取csv文件，然后添加一些列。之后，尝试以orc格式保存。
我不明白火花是如何决定不同阶段任务数量的。
为什么CSV阶段的任务数是1，而ORC阶段的任务数是39？
val c1c8 = spark.read.option("header",true).csv("/user/DEEPAK_TEST/C1C6_NEW/")
val c1c8new = { c1c8.withColumnRenamed("c1c6_F","c1c8").withColumnRenamed("Network_Out","
浏览 5提问于2021-07-21得票数 0
回答已采纳
我在spark中运行reduceByKey。我的程序是spark最简单的例子：
val counts = textFile.flatMap(line => line.split(" ")).repartition(20000).
.map(word => (word, 1))
.reduceByKey(_ + _, 10000)
counts.saveAsTextFile("hdfs://...")
但它总是会耗尽内存...
我使用50台服务器，每台服务器35个执行器，每台服务器14
浏览 0提问于2015-07-01得票数 8
我现在正在笔记本电脑上使用Spark和Scala。
当我将RDD写入文件时，输出将被写入两个文件"part-00000“和"part-00001”。如何强制Spark / Scala向写入一个文件
我的代码是：
myRDD.map(x => x._1 + "," + x._2).saveAsTextFile("/path/to/output")
我在哪里写键，值对。
浏览 4提问于2016-02-17得票数 5
回答已采纳
我找不到多少关于确保分区顺序的文档--我只想确保给定一组确定性转换(输出行总是相同)，如果基础数据集不改变，分区总是接收相同的元素集。这有可能吗？
不需要排序:一个例子是在RDD上应用了一组转换之后，现在的-> (A，B，C，D，E，F，G)
如果我的spark.default.parallelism为2或3，则元素的集合总是：(A，B，C，D)，(E，F，G)或(A，B)，(C，D)，(E，F，G)。
这是因为我必须使我的执行者基于它正在操作的分区/元素集造成一些副作用，我想确保Spark应用程序是幂等的。(如果重新启动，也会产生同样的副作用)
编辑:很明显，DF重新分区是确定性的，但
浏览 5提问于2020-01-10得票数 6
回答已采纳
在我的Spark Streaming作业中，CPU未得到充分利用(仅5-10%)。
它从Kafka获取数据并发送到DynomoDB或thridparty端点。
假设端点不是瓶颈，是否有任何建议的作业可以更好地利用cpu资源。
浏览 1提问于2017-03-18得票数 0
我正在尝试在JMSTemplate方法中使用Spring rdd.foreach类，但是我得到的是任务而不是可序列化的错误。当我尝试使用静态变量时，它在本地运行，但在集群中，我得到的是空指针异常。
示例代码：
inputRDD.foreach(record -> {
messageServices.send(record);
}
错误日志：
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.
浏览 3提问于2015-12-29得票数 0
回答已采纳
2回答
我有一个蜂窝表，只有很少的bigint或字符串列，超过3800万行，总大小略超过1GB，测试环境是一个小的独立集群，有4个工作节点，每个都有8GB内存，Spark1.4。在Spark shell中，我尝试执行SQL。
create table bla as select user_id, brand_id, sum(cnt) from foo group by user_id, brand_id
有几次，工作总是停留在第一阶段，很少有任务挂起。
GC报告似乎表明内存不足，无法存储瞬态对象，进程正在等待完整的GC完成。
一个挂起节点的GC输出：
4572.821: GC [PSYoungGen
浏览 6提问于2015-07-18得票数 2
1回答
调整纱线中的火花作业
、、、
由于java.lang.OutOfMemoryError: Java堆空间，我的火花作业失败了。我试着玩配置参数，如executor-cores，executor-memory，num-executors，driver-cores，driver-memory，spark.yarn.driver.memoryOverhead，spark.yarn.executor.memoryOverhead，根据。下面是我的配置集
--master yarn-cluster --executor-cores 4 --executor-memory 10G --num-executors 30 --driver
浏览 0提问于2018-05-04得票数 0
回答已采纳
2回答
我正在尝试在spark数据集中做一个简单的计数和分组。然而，每次都有一个阶段卡住了(200/2011运行)。我已经重试了几个分区，范围从1000到6000。每次我停留在状态栏中显示(1000/10011运行)或(6000/6001 1运行)的阶段。
请帮助我，这个额外的1任务是从哪里产生的。
spark-submit选项如下：
--conf spark.dynamicAllocation.enabled=false --conf spark.kryoserializer.buffer.max=2000m --conf spark.shuffle.service.enabled=true --c
浏览 0提问于2018-09-27得票数 0
1回答
如何提高火花性能？
、、、、
我有处理大型数据集的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
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
