在Spark中倾斜
是指在数据处理过程中，某些特定的键值对数据量过大或者过小，导致任务的负载不均衡，从而影响整体的计算性能和效率。
倾斜问题在大规模数据处理中是比较常见的，特别是在使用Spark进行分布式计算时。当数据集中某些键的值远远超过其他键的值时，这些键所对应的数据会被分配到同一个分区中，导致该分区的数据量过大，而其他分区的数据量相对较小。这样就会导致负载不均衡，使得计算任务在某些节点上运行时间过长，而其他节点处于空闲状态。
为了解决Spark中的倾斜问题，可以采取以下几种方法：
- 预聚合（Pre-Aggregation）：对于可能导致倾斜的键值对，在计算之前进行预聚合操作，将其合并为一个键值对。这样可以减少倾斜键值对的数量，提高计算的均衡性。
- 随机前缀（Random Prefix）：对于倾斜的键值对，可以在键的前面添加一个随机的前缀，使得键的分布更加均匀。这样可以避免倾斜键值对被分配到同一个分区中。
- 重分区（Repartition）：通过对数据进行重新分区，将倾斜的键值对均匀地分布到不同的分区中，从而实现负载均衡。可以使用Spark提供的repartition或者coalesce方法进行重分区操作。
- 增加并行度（Increase Parallelism）：增加计算任务的并行度，使得倾斜的键值对可以被更多的计算节点同时处理，从而减少计算时间。
- 动态调整分区（Dynamic Partitioning）：根据数据的实际情况，动态调整分区的数量，使得每个分区中的数据量相对均衡。
对于Spark中的倾斜问题，腾讯云提供了一系列的解决方案和产品，如腾讯云的弹性MapReduce（EMR）和弹性数据处理（EDP）等。这些产品提供了自动化的倾斜处理功能，可以根据数据的特点和需求，自动选择合适的处理策略，提高计算的效率和性能。
更多关于腾讯云的倾斜处理产品和解决方案，请参考腾讯云官方文档：腾讯云倾斜处理产品介绍。
相关·内容
1回答
优化火花工作-火花2.1
、、、
我的火花工作目前在59分钟内完成。我想优化它，这样我就可以减少时间。我已经注意到，工作的最后一步需要很多时间(55分钟)(见下面的spark中火花作业的截图)。
我需要加入一个大数据集和一个较小的数据集，在这个连接的数据集上应用转换(创建一个新列)。
最后，我应该基于列PSP对数据集进行重新分区(参见下面的代码片段)。我还在末尾执行排序(根据3列对每个分区进行排序)。
所有细节(基础设施、配置、代码)都可以在下面找到。
我的代码片段：
spark.conf.set("spark.sql.shuffle.partitions", 4158)
val uh = u
浏览 2提问于2019-10-22得票数 0
谁能解释在Apache的生产中如何处理倾斜的数据？
场景：
我们使用"spark-submit"提交了spark作业，并且在spark中观察到很少有任务花费很长的时间，这表明存在偏差。
问题：
(1)我们应采取哪些步骤(重新划分、合并等)？
(2)我们是否需要终止作业，然后在jar中包含倾斜的解决方案并重新提交作业？
(3)能否通过直接从shell运行(合并)之类的命令来解决这个问题，而不杀死作业？
浏览 0提问于2018-06-25得票数 3
在设置FetchFailedException = 2700时，当我加入到表中时，我会遇到一个spark.sql.shuffle.partitions
但是，当设置spark.sql.shuffle.partitions = 500时，可以成功运行。
如我所知，增加shuffle.partitions将减少每次任务中的数据洗牌时。
我错过了什么吗？
异常
FetchFailed(BlockManagerId(699, nfjd-hadoop02-node120.jpushoa.com, 7337, None), shuffleId=4, mapId=59, reduceId=1140, mes
浏览 6提问于2020-06-14得票数 0
1回答
我想问一个数据扭曲的问题，我在我的火花工作。我知道，如果你的数据有偏差，最好的实践技巧之一是做盐渍技术。在我的星星之火工作中，我必须在两个dataframe (其中一个大小在5GB左右)之间连接。
此外，我通常通过一个数字和我用来连接的字段重新分区，以便在进行连接之前尽可能多地控制分区。因此，我将带有混搭的联接转换为映射--具有窄依赖关系的联接。
df.repartition(numRep, colJoin: _*).sortWithinPartitions(colJoin: _*)
设想如下：
我有24个执行者，每个人有4个核心
我把shufflePartitions设为96
我
浏览 1提问于2021-07-19得票数 0
无法将SparkR加载到我的RStudio中。当我尝试加载星火上下文时，我会得到以下错误：
找不到函数“launchBackend”。
我可以让sparkR在我的终端外壳上工作，但这也是因为我不需要加载星火上下文(看起来它已经存在了)。有什么帮助吗？
Sys.setenv(SPARK_HOME="~/spark-1.4.1")
.libPaths(c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib"), .libPaths()))
library(SparkR)
sc <-
浏览 1提问于2015-08-14得票数 3
1回答
Spark有一个可用的倾斜提示(请参见)。是否有类似的提示可用于Spark？
示例这是Spark代码，其中事实表扭曲了ProductId列：
SELECT /*+ SKEW('viewFact', 'ProductId') */
RevSumDivisionName, RevSumCategoryName, CloudAddOnFlag,
SUM(ActualRevenueAmt) AS RevenueUSD, COUNT(*) AS Cnt
FROM viewFact
INNER JOIN viewPMST ON viewFact.Produc
浏览 1提问于2018-10-18得票数 3
回答已采纳
1回答
我的卡夫卡主题中有数以百万计的信息，不幸的是，无论我得到什么时间限制，星火流似乎都在排队等待所有的消息。我已经设置了16核和64 of的独立服务器，我给了我的驱动程序12G和执行器12G内存。
16/09/14 19:27:45 WARN TaskSetManager: Lost task 4.0 in stage 1.0 (TID 7, 10.206.41.172): java.lang.OutOfMemoryError: GC overhead limit exceeded
at com.twitter.chill.Tuple4Serializer.read(TupleSe
浏览 1提问于2016-09-14得票数 1
回答已采纳
我使用Spark将2100万条记录插入到Cassandra表中。spark作业大约需要一个小时，并成功插入所有记录，但在最后阶段(62/63)挂起。
我的Spark属性：
spark.driver.memory 10g
spark.executor.cores 1
spark.executor.instances 40
spark.executor.memory 4g
将记录插入到Cassandra并在最后阶段挂起的代码行-
raw_data_final.rdd.saveToCassandra("marketing_dev1", "offer_detail_11"
浏览 1提问于2018-06-23得票数 0
我同意iterative和interactive编程范式使用spark比map-reduce更好。我也同意我们可以使用HDFS或者任何像HBase这样的hadoop数据存储作为Spark的存储层。
因此，我的问题是-在现实世界中，我们是否有任何用例可以说hadoop MR在这些上下文中比apache spark更好。这里"Better“是用来表示performance, throughput, latency的。与使用spark相比，hadoop MR仍然是进行批处理的好方法。
如果是这样的话，谁能告诉advantages of hadoop MR over apache spark？
浏览 1提问于2015-08-03得票数 3
1回答
我正在亚马逊的EMR上运行一个Spark作业，在客户端模式下，使用吡火花来处理两个大小为200 GB的输入文件中的数据。
作业将数据连接在一起(使用reduceByKey)，执行一些映射和过滤器，并以Parquet格式将其保存到S3中。当作业使用Dataframes进行保存时，所有实际的转换和操作都是在RDDs上执行的。
注意，我已经包含了我的当前配置和值的详细概述，我已经在“故障”部分之后对这些配置和值进行了实验。
代码
与我们所看到的失败相关的代码发生在reduceByKey步骤中。我已经包含了几行上下文，以显示一个先前的map函数和在RDD上实际触发reduceByKey的保存操作：
浏览 1提问于2017-03-28得票数 5
回答已采纳
我的工作是火花1.6，它失败了我的工作与以下错误
/data/05/dfs/dn/yarn/nm/usercache/willir31/appcache/application_1413512480649_0108/spark-local-20141028214722-43f1/26/shuffle__312_0.index (无此类文件或目录)java.io.FileOutputStream.open(原生方法) java.io.FileOutputStream.(FileOutputStream.java:221) org.apache.spark.storage.DiskBlockOb
浏览 3提问于2016-07-14得票数 4
6回答
我发现AWS设置了执行器实例，内存限制为5 Gb --conf spark.executor.memory=5g，有时在一个大数据集中，它会在java.lang.OutOfMemoryError中失败。驱动程序实例--spark.driver.memory=5g也是如此。是否有任何增加此值的选择？
浏览 0提问于2018-02-28得票数 18
回答已采纳
1回答
我面临一些数据倾斜的问题，我想看看火花提示是否可以帮助我。但是，我不确定是否支持倾斜提示。这些站点： 都暗示支持它们。
然而，声称Spark SQL supports COALESCE and REPARTITION and BROADCAST hints. All remaining unresolved hints are silently removed from a query plan at analysis.，这意味着我的倾斜提示也将被忽略。
哪一个是对的？我在运行Spark2.2
浏览 0提问于2018-11-29得票数 1
回答已采纳
我试图根据执行器和内核的数量来衡量Spark中的性能。我们的想法是玩：
spark.conf.set("spark.executor.instances", "x")
spark.conf.set('spark.cores.max', 'x')
当我改变执行器和核心的数量时，测试Spark的性能的影响。数据是1.66 hp的推特文件.json我使用的是一台计算机hp：
处理器:英特尔(R)酷睿(TM) i7-8650U CPU @1.90 GB 2.11 GB // 16 GB RAM
import time
st
浏览 0提问于2019-02-07得票数 1
1回答
编辑：注释:执行器通常会发出消息[GC (Allocation Failure) ]。它运行它是因为它试图将内存分配给执行器，但是执行器已经满了，所以当向执行器加载新的东西时，它将尝试腾出空间。如果您的执行者在一个循环中这样做，这可能意味着您试图加载到该执行器中的内容太大了。
我在AWS EMR 5.8.0上运行Spark2.2，Scala2.11
我试图在拒绝完成的数据集中运行一个count操作。令人沮丧的是，它只挂在一个特定的文件上。我在与S3不同的文件上运行这个作业，没问题--它完全完成了。原始CSV文件本身是@18 to，我们对它运行一个转换，将原来的CSV转换为一个struct列，给
浏览 2提问于2019-10-01得票数 0
回答已采纳
1回答
我正在用火花壳执行一个火花-scala任务，而我面临的问题是，在最后阶段的末尾和最后的映射器，就像在第5阶段，它分配了50个，完成了49个，在第50阶段，它需要5分钟的时间，然后说出内存，然后失败。我正在使用SPARK_MAJOR_VERSION=2
我正在使用下面的命令spark-shell --master yarn --conf spark.driver.memory=30G --conf spark.executor.memory=40G --conf spark.shuffle.service.enabled=true --conf spark.dynamicAllocation.e
浏览 0提问于2019-04-26得票数 0
回答已采纳
3回答
我正在像下面这样写我的数据
df.write().format("com.databricks.spark.avro").save("path");
但是，我得到了大约200个文件，其中大约有30-40个文件是空的，我可以理解这可能是由于空分区造成的。然后，我更新了代码如下
df.coalesce(50).write().format("com.databricks.spark.avro").save("path");
但我觉得这可能会影响表演。还有其他更好的方法来限制输出文件的数量和删除空文件吗？
浏览 1提问于2017-05-18得票数 0
回答已采纳
我是新手，对Java没有编程经验。我正在使用pyspark处理一个非常大的时间序列数据集，其中有近4000列数值(浮点)列和数十亿行。
我想用这个数据集实现以下目标：
时间序列数据的间隔为10毫秒。我希望按1s间隔对数据进行分组，并使用平均值作为聚合函数。
这里是我用来读取分区拼花文件的代码.
df = (spark.read.option("mergeSchema", "true")
.parquet("/data/"))
这里是我编写的用于groupby和聚合的代码：
col_list = [... list of nu
浏览 2提问于2019-10-03得票数 1
1回答
在将数据(假设表)传输到HDFS之后，我不知道如何复制数据(哪一部分属于哪个machne (节点))。
因此，运行Spark查询一些人说，您可以提示我的数据是倾斜的。
但是，我怎么知道我的数据是扭曲的，这样我就能给出火花的提示？
浏览 2提问于2018-06-27得票数 2
回答已采纳
2回答
我有一个性能问题，在分析了Spark之后，我发现了数据的倾斜性：
一开始我认为分区分布不均匀，所以我分析了每个分区的行数，但这似乎很正常(没有异常值)：
但是问题仍然存在，我看到有一个执行者处理大部分数据：
因此，现在的假设是，分区在执行程序中的分布并不均匀，问题是：如何将火花分区分配给执行者？和如何更改它以解决我的偏斜问题？
代码非常简单：
hive_query = """SELECT ... FROM <multiple joined hive tables>"""
df = sqlContext.sql(hive
浏览 1提问于2021-04-16得票数 3
回答已采纳
我为协作过滤推荐系统编写了一个spark程序(Python3.6和Spark2.3.2)，该程序适用于2种情况：
Case 1:基于项的CF推荐系统Case 2:基于Min-Hash LSH的基于用户的CF推荐系统
我已经写了，训练和预测程序，有这两个案例。我的代码适用于基于用户的推荐，但是当我试图训练基于项的CF模型时，我会得到以下错误：
2020-10-18 20:12:33 ERROR Executor:91 - Exception in task 0.0 in stage 23.0 (TID 196)
org.apache.spark.api.python.PythonExceptio
浏览 6提问于2020-10-19得票数 1
专家们，我注意到了一件奇怪的事情，那就是生产中的一项工作(在纱线集群模式下运行)。在执行了大约一个小时+(大约65-75分钟)之后，它就会消失，而不会抛出任何特定的错误消息。我们已经分析了大约2周的纱线日志，没有特别的错误，它只是在进行ETL操作(读/写蜂窝表、做简单的地图、修剪、lambda操作等)时死在中间，而不是任何要指出的特定代码。有时，重新运行修复它，有时它需要多次重新运行。代码是优化的，火花提交-conf拥有所有正确优化的选项。正如我们前面提到的，它运行的非常完美，大约有30个其他应用程序具有非常好的性能统计。这些都是我们的选择-
spark-submit --conf spark
浏览 8提问于2019-10-25得票数 1
回答已采纳
我有一个分类任务。我想用apache spark ml lib SVM算法进行分类。我有n维的输入数据。在特征向量中，可能缺少一些维度。
如何处理缺失值？我认为假设缺失值为零或其他值是错误的。
浏览 3提问于2015-10-02得票数 1
1回答
我有一个没有Serverless的应用程序，由于某些原因，它会陷入执行超时。我已经测试了所有的s3连接，而且它正在工作。此问题发生在spark表中查询的执行过程中。
EMR版本是: emr-6.7.0 --在k8s中SPAR3.1.1版本上运行相同的任务是不可能的，可能是与版本相关的。
我的火花会话设置：
spark = (SparkSession.builder
.config("spark.hadoop.fs.s3a.fast.upload", True)
.config("spark.hadoop.fs.s3a.impl"
浏览 35提问于2022-09-28得票数 0
使用Spark rdd有一种方法来groupByKey，然后在每个组中排序，然后映射到大型数据集。天真的方法是在每个组上映射，并为每个组创建一个列表并对其进行排序。但是，这种列表的创建可能会导致具有许多条目的组出现内存不足问题。有没有办法让Spark进行排序，以避免内存不足的问题。
浏览 0提问于2021-09-07得票数 0
我有一个有100个分区的df文件，在保存到HDFS作为.parquet之前，我想减少分区的数量，因为拼图文件太小(<1MB)。我在写之前添加了coalesce：
df.coalesce(3).write.mode("append").parquet(OUTPUT_LOC)
它可以工作，但会将过程从每个文件2-3秒减慢到每个文件10-20秒。当我尝试重新分区时：
df.repartition(3).write.mode("append").parquet(OUTPUT_LOC)
这个过程一点也不慢，每个文件2-3秒。
为什么？当减少分区数量时，联合不应该总是
浏览 16提问于2021-05-20得票数 1
我们使用Apache Beam，它是在火星车上执行的。我们的案件如下。这两个用例都会导致OutofMemory错误。
1)连接-2个使用Apache的大表--一个大小为120 is的表，另一个为60 is的表。这会在OutofMemory ()内部调用groupByKeyOnly()时导致GroupCombineFunctions.java错误。
2) GroupByKey --我们根据如下键对数据集进行分组。PCollection>> costBasisRecords = masterDataResult.apply(GroupByKey.create())；
此GroupbyK
浏览 1提问于2017-12-16得票数 0
回答已采纳
我们有大约4000万条记录用于表。
如何在使用Spark 1.6.2从Oracle读取数据时选择lowerBound、upperBound和numPartitions的值。
我有分区键。我们有datastax群集-3个节点-每个节点18个核心-每个节点27 GB
提前谢谢。
浏览 1提问于2018-02-02得票数 3
回答已采纳
我必须首先按“客户组”进行分区，但我也希望确保每个"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
回答已采纳
我在pyspark中运行一个作业，在那里我曾经使用过一个。这会导致以下错误(此处简称)：
org.apache.arrow.vector.util.OversizedAllocationException: Unable to expand the buffer
我很确定这是因为熊猫UDF收到的其中一个组是巨大的，如果我减少数据集并删除足够多的行，我就可以运行我的UDF，没有任何问题。但是，我想使用我的原始数据集运行，即使我在一台拥有192.0 GiB内存的机器上运行这个火花作业，我仍然会得到相同的错误。(而且192.0 GiB应该足够在内存中保存整个数据集。)
我如何给火花足够的内存，以便能
浏览 31提问于2019-10-07得票数 4
回答已采纳
2回答
我们有相当少的问题，星火节约服务器
从日志中我们可以看到：未能将RPC 9053901149358924945发送到/DATA节点计算机:50149
请建议为什么会发生这种情况，以及解决办法是什么？
Failed to send RPC 9053901149358924945 to /DATA NODE MACHINE:50149: java.nio.channels.ClosedChannelException
more spark-hive-org.apache.spark.sql.hive.thriftserver.HiveThriftServer2-1-master03.sys67.c
浏览 2提问于2018-02-07得票数 3
如果我的卡夫卡主题收到这样的记录
CHANNEL | VIEWERS | .....
ABC | 100 | .....
CBS | 200 | .....
我有Spark结构化的流媒体代码来读取和处理Kafka记录如下：
val spark = SparkSession
.builder
.appName("TestPartition")
.master("local[*]")
.getOrCreate()
import spark.implicits._
浏览 0提问于2018-04-20得票数 5
我们已经使用Spark 1.6.1和Scala2.10编写了一些代码来通过GraphX连接和链接数据集。
如果我们在spark-shell中运行这段代码，它将在30分钟内完成，具有10个执行器，10g内存和每个执行器5个cpu。
如果我们使用spark-submit在一个fat jar中运行它，那么这个过程就会因为内存不足错误而失败，并且需要一个半小时才能到达那个点。
有没有人知道这可能是什么原因？
有谁知道SparkContext和SQLContext是如何在shell中设置的，以及我们在设置自己的上下文时是否遗漏了什么，这可能会导致spark-submit运行得如此糟糕？
浏览 18提问于2017-02-02得票数 1
回答已采纳
1回答
在我的Scala/Spark应用程序中，我试图正确地使用多处理。从下面的代码中可以看到，线程数等于storage数组中的元素数。我测试了当前的代码，它可以工作。但是如您所见，storage数组中只有两个元素。在我看来，如果数组中有大量的元素，就会出现问题。在我的例子中，我不知道将来数组中会有多少个元素。也许我应该限制线程的数量，只有在处理之前的线程时才启动新的线程。
问题：如何确定最佳线程数？
Main.app:
import org.apache.spark.sql.DataFrame
import utils.CustomThread
object MainApp {
def m
浏览 3提问于2020-01-29得票数 0
我在kubernetes上执行一项星火作业，随着大量的数据，我经常得到“执行者丢失”，执行者被杀死，任务失败。我已经在所有运行的执行器吊舱上做了一个kubectl logs -f，但是我从来没有看到任何异常被抛出(我希望类似于OutOfMemoryError之类的东西)。豆荚会突然停止计算，然后被直接移除，因此它们甚至不会停留在Error状态，从而无法挖掘和排除故障。他们就这么消失了。
我应该如何解决这个问题？在我看来，库伯奈特斯自己杀死了豆荚，因为我认为它们超越了某些界限，但据我所知，豆荚应该处于Evicted状态(或者它们不应该？)
这似乎与内存的使用有关，因为当我打开spark.exec
浏览 1提问于2020-08-18得票数 5
回答已采纳
我正在尝试配置一个spark作业，在Dataproc集群上使用固定资源运行，但是在作业运行6分钟后，我注意到除了7个执行器之外，所有的执行器都被删除了。45分钟后，作业完全没有进展，我找不到任何错误或日志来解释。 当我检查作业详细信息中的时间线时，它显示除了7个执行器之外的所有执行器都在6分钟标记时被删除，并显示消息Container [really long number] exited from explicit termination request.。 我正在运行的命令是： gcloud dataproc jobs submit spark --region us-central1
浏览 39提问于2021-04-22得票数 1
1回答
有火花的任务调度
、、、、
我正在我的4节点集群上运行相当大的任务。我从一个表中读取了大约4GB的过滤数据，并运行Na ve的训练和预测。我让HBase区域服务器运行在一台机器上，这台机器与spark集群分离，运行在公平的调度模式下，尽管HDFS在所有机器上运行。
在执行过程中，在集群上活动任务的数量方面，我遇到了奇怪的任务分布。我观察到，在任何时候，只有一个活动任务或最多两个任务在一个/两个机器上运行，而另一个任务处于空闲状态。我的期望是，RDD中的数据将在所有节点上被分割和处理，以便进行计数和不同的操作等。为什么没有将所有节点用于单个作业的大型任务？在另一台机器上安装HBase与此有任何关系吗？
浏览 1提问于2014-09-29得票数 3
回答已采纳
我正在使用Spark Dataframe中的numPartitions，lowerBound，upperBound从oracle.but中获取大表，它给我的数据只相当于一个分区。假设分区数为10，记录总数为100，那么我在Hive中只能摄取10条记录(记录总数/numPartitions)。下面是我的代码片段
val hiveContext = SparkApp.getHiveContext("AppName")
val jdbcUsername = "MYUSERNAME"
val jdbcPassword = "MYPASSWORD"
浏览 0提问于2018-07-02得票数 1
我想处理一个250 Hd的gzip(filename.json.gzip)文件，在Azure集群中使用Spark。但我做不到。
我想是因为内核、ram、vCPU之间的关系不好，所以我想创建更好的集群和发送火花配置。
目前，我正在使用这个实例：
集群E8a v4的6个节点(8核，64 GB内存)
我的星火配置是：
驱动程序内存:10
驱动核心:7
执行器内存:10
执行者核心:7
遗嘱执行人:7人
那么，在Azure HDInsight集群()和火花子配置中还有一个更好的选择吗？
浏览 3提问于2021-09-09得票数 0
我想深入了解合并排序加入星火的概念。我理解总体思路:这是与合并排序算法相同的方法:获取2个排序数据集，比较第一行，编写最小行，重复。我还理解如何实现分布式合并排序。
但是，我无法了解它是如何在Spark中实现的，涉及分区和执行器的概念。
，这是我的想法，
考虑到我需要连接两个表A和B。如果这很重要的话，表可以通过Spark从Hive读取。
默认情况下，Spark使用200个分区。
火花将计算连接键范围(从minKey(A，B)到maxKey(A，B) )，并将其分割成200个部分。这两个数据集将按键范围划分为200个部分:a-分区和B-分区.
每个与相同密钥相关的A-分区和B-
浏览 5提问于2021-04-29得票数 6
6回答
我正在使用Spark RDD加入两个大型数据集。一个数据集非常不对称，所以很少有执行器任务需要很长时间才能完成。我该如何解决这个场景？
浏览 3提问于2016-11-02得票数 29
2回答
我有一个火花作业，运行以下代码的文件。但是，此步骤会在输出文件夹中创建几个文件。
sampledataframe.write.mode('append').partitionBy('DATE_FIELD').save(FILEPATH)
因此，我已经开始使用下面这行代码来重新划分这些文件，并拥有一个单独的文件。
sampledataframe.repartition('DATE_FIELD').write.mode('append').partitionBy('DATE_FIELD').save(FILEPATH)
浏览 5提问于2019-09-27得票数 0
1回答
我正在使用PySpark和Hive开发一个星火集群。
我在这里看到了很多问题，所以关于“不能覆盖”也是从“Hive错误”读取的表。我明白这来自于蜂巢本身的工作方式，我们正计划为此搬到库都:但现在我需要完成与蜂巢的工作。我找到了一些解决方案，几乎所有的解决方案都依赖于“打破数据谱系”原则：
对临时表进行tablecheckpoint 保存，然后将数据存储在右上。
和类似的。我不想使用它们，主要是因为它们在大型数据集上效率不高，所以我想出了如下方法：
bananaDF = spark.sql("select * from banana")
// hundreds of trasfo
浏览 0提问于2020-08-25得票数 0
我的日常火花应用程序处理所有的数据采集步骤，从处理原始层(去重复和其他功能)开始，创建数据仓库和数据仓库。当我启动spark.sql.shuffle.partitions值时，我无法决定什么是更好的sparkSession值配置。我的集群配置：
7个活动节点
738.50 GB内存总计
火花版本2.1.0
我的问题是:如何在我的火花工作中选择更好的spark.sql.shuffle.partitions价值来获得性能。谢谢
浏览 4提问于2020-11-18得票数 0
在Spark中加入大型数据格式时，我得到了与相同的错误。建议设置MEMORY_AND_DISK和/或spark.shuffle.memoryFraction 0。但是，在Spark 1.6.0中不推荐使用spark.shuffle.memoryFraction，如果我没有缓存任何>=或Dataframe，那么设置MEMORY_AND_DISK就不会有帮助了，对吧？此外，我还得到了许多其他警告日志和任务重试，这使我认为任务是不稳定的。
因此，我的问题是：
在Spark >= 1.6.0中加入大型数据格式的最佳实践是什么？
更具体的问题是：
如何调整执行程序的spark
浏览 2提问于2016-06-23得票数 12
回答已采纳
我们有一个要求，从一个非分区的work_db.customer_tbl外部蜂窝表通过PySpark从一个分区的外部的蜂巢表通过PySpark摄取数据，以前是通过hive查询完成的。最后一个表由列 load_date (load_date列的格式为load_date)进行分区。
因此，我们有一个简单的PySpark脚本，它使用一个插入查询(与前面使用的单元查询相同)来使用spark.sql()命令摄取数据。但是，我们有一些严重的性能问题，因为我们在摄入后试图摄入的表大约有3000分区，和每个分区大约有4MB的数据，除了最后一个分区大约4GB。总表大小接近15 is。另外，在摄入每个分区之后，每
浏览 2提问于2021-03-25得票数 0
3回答
我在Scala中有一个程序，它读取一个CSV文件，向Dataframe添加一个新列，并将结果保存为一个parquet文件。它在小文件(<5 Go)上工作得很好，但是当我尝试使用更大的文件(~80 Go)时，它总是会失败，因为它应该用这个堆栈跟踪来编写拼花文件：
16/10/20 10:03:37 WARN scheduler.TaskSetManager: Lost task 14.0 in stage 4.0 (TID 886, 10.0.0.10): java.io.EOFException: reached end of stream after reading 136445 by
浏览 0提问于2016-10-20得票数 1
1回答
我有一个基本的火花工作，做几个连接。加入的3个数据框架有点大，每一个都有近20亿条记录。我有一个火花基础设施，可以在必要时自动扩展节点。这似乎是一个非常简单的seems查询，我将其结果写入磁盘。但是当我从Spark中看到时，这份工作总是停留在99%的水平上。
我尝试过的很多事情是：
增加executors和executor memory的数量。
在编写文件时使用repartition。
使用本机火花join而不是spark SQL join等
然而，所有这些都没有起作用。如果有人能分享解决这个问题的经验，那就太好了。提前谢谢。
浏览 1提问于2017-11-13得票数 3
我的任务是对两个表/星火数据执行窗口聚合(即根据表2中的输入日期t-，回顾表1中的t-x)。
目前，我已经想出了两种不同的方法来解决这个问题--要么是：
在窗口aggregationCarry之前，在两个数据之间执行outer join (生成A行)，然后在两个数据(生成B行，其中B<代码>E 215</code> is ><代码>E 116</代码>A<代码>E 217</代码>)之前，在窗口聚集之前执行unionByName (通常为1.5- 1.75x)，其中B</代码></代码>G 2
浏览 5提问于2022-10-15得票数 1
回答已采纳
2回答
我为我的星火作业启用了Kryo序列化，启用了该设置要求注册，并确保我的所有类型都已注册。
val conf = new SparkConf()
conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.set("spark.kryo.registrationRequired", "true")
conf.registerKryoClasses(classes)
conf.registerAvroSchemas(avroSchema
浏览 5提问于2017-01-09得票数 16
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
