如何在Spark scala中对倾斜列上的数据帧进行重新分区？
在Spark Scala中对倾斜列上的数据帧进行重新分区的方法是使用Spark的自定义分区器来解决倾斜问题。下面是一个完整的解决方案：
- 首先，我们需要确定哪一列是倾斜列。倾斜列是指数据分布不均匀，导致某些分区的数据量远大于其他分区的列。
- 接下来，我们可以使用Spark的
groupBy操作将数据按照倾斜列进行分组。
val groupedData = dataFrame.groupBy("skewedColumn")- 然后，我们可以使用
count函数计算每个分组的数据量，并找到数据量最大的分组。
val skewedGroup = groupedData.count().orderBy(desc("count")).limit(1).collect()(0)(0)- 接下来，我们可以使用
filter函数将数据分为倾斜分区和非倾斜分区。
val skewedData = dataFrame.filter(dataFrame("skewedColumn") === skewedGroup)
val nonSkewedData = dataFrame.filter(dataFrame("skewedColumn") =!= skewedGroup)- 然后，我们可以使用
repartition函数对倾斜分区进行重新分区。
val repartitionedSkewedData = skewedData.repartition(numPartitions)其中，numPartitions是重新分区的数量，可以根据实际情况进行调整。
- 最后，我们可以将倾斜分区和非倾斜分区合并为一个新的数据帧。
val result = repartitionedSkewedData.union(nonSkewedData)这样，我们就成功地对倾斜列上的数据帧进行了重新分区。
对于这个问题，腾讯云提供了适用于Spark的弹性MapReduce（EMR）服务，可以帮助用户快速搭建和管理Spark集群，提供高性能的计算和存储能力。您可以通过以下链接了解更多关于腾讯云EMR的信息：腾讯云EMR
请注意，以上答案仅供参考，具体实现方法可能因实际情况而异。
相关·内容
我正在尝试修复在我的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
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
回答已采纳
我们有一个要求，从一个非分区的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
我使用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
在设置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
我使用以下命令在本地模式下使用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
回答已采纳
我的工作是火花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
我正在为确定父子表的表生成层次结构。
下面是使用的配置，即使在收到与太大的帧有关的错误之后：
火花特性
--conf spark.yarn.executor.memoryOverhead=1024mb \
--conf yarn.nodemanager.resource.memory-mb=12288mb \
--driver-memory 32g \
--driver-cores 8 \
--executor-cores 32 \
--num-executors 8 \
--executor-memory 256g \
--conf spark.maxRemoteBlockSizeFetc
浏览 3提问于2018-07-11得票数 8
我们使用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
回答已采纳
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回答
我想问一个数据扭曲的问题，我在我的火花工作。我知道，如果你的数据有偏差，最好的实践技巧之一是做盐渍技术。在我的星星之火工作中，我必须在两个dataframe (其中一个大小在5GB左右)之间连接。
此外，我通常通过一个数字和我用来连接的字段重新分区，以便在进行连接之前尽可能多地控制分区。因此，我将带有混搭的联接转换为映射--具有窄依赖关系的联接。
df.repartition(numRep, colJoin: _*).sortWithinPartitions(colJoin: _*)
设想如下：
我有24个执行者，每个人有4个核心
我把shufflePartitions设为96
我
浏览 1提问于2021-07-19得票数 0
我有一个数据，并写入到S3桶目标位置。在代码中，合并用于加载数据和获取SparkOutOfMemoryError.The当前代码，这些代码使用了多个项目，并且看到了许多推荐的重新分区解决方案，它对我起了作用。即使它没有记录，合并也不起作用。有没有其他的方法来解决这个问题而不改变为重新划分？
代码：
empsql = 'Select * From Employee'
df = spark.sql(empsql) ##Spark is configured
df.coalesce(2).write.mode('overwrite').format("parq
浏览 8提问于2020-06-11得票数 2
df.write.partitionBy("par").format("orc").saveAsTable("mytable")
大家好，当我将spark数据帧保存为分区的hive表时，过程非常非常慢，有人知道为什么吗？是否有需要调整的参数？
浏览 0提问于2017-09-15得票数 2
1回答
优化火花工作-火花2.1
、、、
我的火花工作目前在59分钟内完成。我想优化它，这样我就可以减少时间。我已经注意到，工作的最后一步需要很多时间(55分钟)(见下面的spark中火花作业的截图)。
我需要加入一个大数据集和一个较小的数据集，在这个连接的数据集上应用转换(创建一个新列)。
最后，我应该基于列PSP对数据集进行重新分区(参见下面的代码片段)。我还在末尾执行排序(根据3列对每个分区进行排序)。
所有细节(基础设施、配置、代码)都可以在下面找到。
我的代码片段：
spark.conf.set("spark.sql.shuffle.partitions", 4158)
val uh = u
浏览 2提问于2019-10-22得票数 0
在使用mllib RandomForest训练数据时，我得到了一个错误。因为我的数据集很大，并且默认分区相对较小。因此抛出的异常指示“大小超过Integer.MAX_VALUE”，原始堆栈跟踪如下所示，
15/04/16 14:13:03警告scheduler.TaskSetManager:阶段6.0中丢失的任务19.0 (TID 120，10.215.149.47)：java.lang.IllegalArgumentException:大小超过Integer.MAX_VALUE
在sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:828
浏览 5提问于2015-04-17得票数 16
回答已采纳
我有一段创建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
我使用的是spark 1.6.1，我正在尝试将数据帧保存为orc格式。
我面临的问题是，保存方法非常慢，在每个执行器上保存50M orc文件大约需要6分钟。这就是我保存数据帧的方式
dt.write.format("orc").mode("append").partitionBy("dt").save(path)
我尝试使用saveAsTable到同样使用orc格式的hive表，这似乎快了20%到50%，但这种方法有其自身的问题-似乎当一个任务失败时，重试总是会因为文件已经存在而失败。这就是我保存数据帧的方式
dt.write.format(&#
浏览 0提问于2016-07-23得票数 2
我在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
我正在尝试在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
回答已采纳
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回答
我正在用火花壳执行一个火花-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
回答已采纳
我正在读取一个巨大的csv文件，包括39,795,158条记录，并在Azure Databricks上写入MSSQL服务器。Databricks(笔记本)运行在一个具有56 GB内存、16个核心和12个工作人员的集群节点上。
这是我在Python和PySpark中的代码：
from pyspark.sql import *
from pyspark.sql.types import *
from pyspark.sql.functions import *
from time import sleep
url = "jdbc:sqlserver://{0}:{1};database=
浏览 10提问于2022-02-11得票数 0
当我通过rdd.repartition(1).saveAsTextFile(file_path)保存一对rdd时，会遇到一个错误。
Py4JJavaError: An error occurred while calling o142.saveAsTextFile.
: org.apache.spark.SparkException: Job aborted.
at org.apache.spark.internal.io.SparkHadoopWriter$.write(SparkHadoopWriter.scala:100)
at org.apache.spark.rdd.
浏览 1提问于2020-07-01得票数 0
回答已采纳
我试图计算负样本的数量，如下所示：
val numNegatives = dataSet.filter(col("label") < 0.5).count
但我得到的大小超过了Integer.MAX_VALUE错误：
java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:869)
at org.apache.spark.storage.DiskStore$$anonfu
浏览 0提问于2018-04-13得票数 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
我在EMR (纱线)上使用SparkSQLv2.4.7。我编写Spark查询来执行转换。
估计复杂查询的最优混叠分区数
我试图估计需要设置的最优洗牌分区的数量，以便获得具有多个联接的复杂查询的最佳性能。在互联网上，我发现分区的最佳大小应该在10 MB-100 MB的范围内。现在，由于我知道这个值，下一步是计算查询的数据洗牌卷(以MB为单位)，然后用100除以获得洗牌分区号。但是，对于涉及多个大表连接的复杂查询，估计洗牌量变得非常困难。
那么，如何估计洗牌卷的数量，以及大查询所需的最优洗牌分区数量？目前(经过大量搜索)，我正在执行以下步骤-
scala> spark.sql("&#
浏览 8提问于2022-01-30得票数 0
谁能解释在Apache的生产中如何处理倾斜的数据？
场景：
我们使用"spark-submit"提交了spark作业，并且在spark中观察到很少有任务花费很长的时间，这表明存在偏差。
问题：
(1)我们应采取哪些步骤(重新划分、合并等)？
(2)我们是否需要终止作业，然后在jar中包含倾斜的解决方案并重新提交作业？
(3)能否通过直接从shell运行(合并)之类的命令来解决这个问题，而不杀死作业？
浏览 0提问于2018-06-25得票数 3
所以，我有如下数据，
[ (1, data1), (1, data2), (2, data3), (1, data4), (2, data5) ]
我想将其转换为以下内容，以供进一步处理。
[ (1, [data1, data2, data4]), (2, [data3, data5]) ]
我使用了groupByKey和reduceByKey，但是由于数据量很大，它失败了。数据不是很高，而是很宽。换句话说，键在1到10000之间，但是值列表从100 K到900 k不等。
我正在努力解决这个问题，并计划应用mapPartitions或(Hash)partitioner。
所以，如果其中一个可行
浏览 5提问于2016-01-26得票数 6
我正在尝试使用Spark在S3事件上创建一个简单的sql查询。我正在加载大约30 as的JSON文件，如下所示：
val d2 = spark.read.json("s3n://myData/2017/02/01/1234");
d2.persist(org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK);
d2.registerTempTable("d2");
然后，我尝试将查询结果写入文件：
val users_count = sql("select count(distinct data.use
浏览 1提问于2017-02-15得票数 26
回答已采纳
我正在使用RandomForestCLassifier构建一个模型。这是我的密码
conf = SparkConf()
conf.setAppName('spark-nltk')
sc = SparkContext(conf=conf)
sqlContext = SQLContext(sc)
m=sc.textFile("Question_Type_Classification_testing_purpose/data/TREC_10.txt").map(lambda s: s.split(" ",1))
df= m.toDF()
创建的data
浏览 6提问于2017-04-01得票数 3
回答已采纳
我在一个文件夹里有几个不同国家的文件。如下所示
Casedata_GBR_202110_timestamp.csv
Casedata_ARG_202110_timestamp.csv
现在，我必须处理这些文件，按国家/地区处理它们，然后复制到各自的文件夹中。我的目标文件夹结构如下所示
2021-->11-->GBR
2021-->11-->ARG
在spark scala/scala中，帮助我编写代码来处理国家/地区的文件并移动到相应的国家/地区文件夹。
浏览 0提问于2021-11-24得票数 0
我正在尝试从mysql中获取一个相当大的表，这样我就可以使用spark/databricks进行操作。我不能让它加载到spark中--我已经尝试了更小的子集，但即使在最小的合理单元，它仍然无法加载。
我尝试过在mysql中使用wait_timeout和interactive_timeout，但似乎没有什么不同
我还加载了一个较小的(不同的)表，并且加载得很好。
df_dataset = get_jdbc('raw_data_load', predicates=predicates).select('field1','field2', 'f
浏览 2提问于2018-01-08得票数 1
我可以使用以下代码从Master节点中的Oracle数据库中读取数据：
val spark = SparkSession
.builder
.master("local[4]")
.config("spark.executor.memory", "8g")
.config("spark.executor.cores", 4)
.config("spark.task.cpus",1)
浏览 1提问于2021-10-16得票数 2
2回答
我正在尝试对消费者群体进行实验。
这是我的代码片段
public final class App {
private static final int INTERVAL = 5000;
public static void main(String[] args) throws Exception {
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put("bootstrap.servers", "xxx:9092");
kafka
浏览 0提问于2018-05-27得票数 2
回答已采纳
1回答
当我执行以下命令时：
scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4).partitionBy(new HashPartitioner(10)).persist()
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[10] at partitionBy at <console>:22
scala> rdd.partitions.size
res9: Int = 10
scala> rdd.partitioner.isDefined
浏览 1提问于2015-12-28得票数 20
回答已采纳
我正在尝试从多个路径读取一些avro文件到DataFrame。假设我的路径是这个路径下的"s3a://bucket_name/path/to/file/year=18/month=11/day=01"，我还有两个分区，假设是country=XX/region=XX
我想一次读取多个日期，而不需要明确命名国家和地区分区。此外，我希望国家和地区是这个DataFrame中的列。
sqlContext.read.format("com.databricks.spark.avro").load("s3a://bucket_name/path/to/file/yea
浏览 2提问于2018-12-03得票数 2
我得到了几个Avro文件，它们是从java类A的对象生成的。使用spark-shell (Spark 1.4.0)，我可以使用spark-avro (2.0.1版)将这些文件读取到数据帧中。
val df = sqlContext.read.avro("file.avro")
只要spark shell不知道类A，就可以正常工作。如果我向shell添加一个jar，并发出相同的命令，我会得到以下异常：
A cannot be cast to org.apache.avro.generic.IndexedRecord
at org.apache.avro.generic.G
浏览 1提问于2015-09-10得票数 1
我正在使用Google Cloud Dataproc做spark工作，我的编辑器是Zepplin。我正在尝试将json数据写入gcp存储桶。当我尝试10MB文件时，它成功了。但由于文件为10 But而失败。我的dataproc有一个主服务器，4CPU，26 My内存，500 My磁盘。5个具有相同配置的工作进程。我猜它应该能够处理10 it的数据。
我的命令是toDatabase.repartition(10).write.json("gs://mypath")
错误是
org.apache.spark.SparkException: Job aborted.
at org
浏览 4提问于2019-04-08得票数 11
回答已采纳
我必须首先按“客户组”进行分区，但我也希望确保每个"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
回答已采纳
当我在一个80节点的集群上运行Azure HDInsight 3.6时，我在Spark 2.3中遇到了这个异常： java.lang.UnsupportedOperationException: Can not build a HashedRelation that is larger than 8G
at org.apache.spark.sql.execution.joins.LongToUnsafeRowMap.grow(HashedRelation.scala:623)
at org.apache.spark.sql.execution.joins.LongToUnsafeRowMa
浏览 44提问于2020-07-02得票数 0
回答已采纳
各位，
当我试图在每行一个键标识符上加入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中运行一个应用程序，它可以在两个数据帧之间进行简单的比较。我在集群环境中以jar文件的形式执行。我的集群环境是94节点集群。有2 GB和4 GB两个数据集映射到数据帧。
我的工作是很好地工作非常小的文件…
我个人认为saveAsTextFile在我的应用程序中花费了更多的时间，在我的集群连接细节下面
Total Vmem allocated for Containers 394.80 GB
Total Vmem allocated for Containers 394.80 GB
Total VCores allocated for Containers 3
浏览 0提问于2017-09-13得票数 1
在解决了这个问题之后：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
我使用以下代码在星火GraphX中创建了一个图形。(见)
import scala.math.random
import org.apache.spark._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import scala.util.Random
import org.apache.spark.HashPartitioner
object SparkER {
val nPartitions: Integer = 4
val n: Long = 100
val p: Double = 0.
浏览 0提问于2015-12-22得票数 3
回答已采纳
1回答
我正面临一个问题，我的火花工作被困在当地，而运行在IntelliJ的想法。我的工作一直持续到一个阶段，比如完成200个任务中的199个，或者完成3个任务中的一个，然后被困在那里。
我试图在IDE中使用评估表达式查看正在发生的事情，并注意到了一个奇怪的问题。如果我使用的是myDf.rdd.map(r => r).cache()
java.io.IOException: Class not found
at org.apache.xbean.asm5.ClassReader.a(Unknown Source)
at org.apache.xbean.asm5.ClassReader.<
浏览 1提问于2018-03-15得票数 1
1回答
我希望有人能帮我解决火花问题。
我在一个数据集(ds_ids)中有一堆I，假设我的数据集如下所示：
ID
1
2
3
我正在收集这些in的进一步数据，并将它们存储在一个新的数据集(ds_combined_data)中，如下所示：
ID | Date | Status
1 | 10.01.18 | 10
1 | 11.01.18 | 20
2 | 10.01.18 | 20
3 | 12.01.18
浏览 1提问于2018-08-01得票数 0
我是scala的新手，我不知道如何问这类问题(技术术语.)。我有个数据：
id VehicleID Longitude Latitude Date Distance
1 12311 55.55431 25.45631 01/02/2020 20
2 12311 55.55432 25.45634 01/02/2020 80
3 12311 55.55433 25.45637 02/
浏览 2提问于2020-03-26得票数 1
回答已采纳
我使用的是Spark 1.6，正在尝试将csv (或tsv)文件读取为数据帧。以下是我采取的步骤：
scala> val sqlContext= new org.apache.spark.sql.SQLContext(sc)
scala> import sqlContext.implicits._
scala> val df = sqlContext.read
scala> .format("com.databricks.spark.csv")
scala> .option("header", "true")
sc
浏览 12提问于2016-07-27得票数 2
我正在尝试用可能在count_2列中的数据替换count_1列中的所有空数据。下面是给定输入的预期输出。我如何在Spark Scala中做到这一点？ 输入数据帧 name count_1 count_2
Java 10000 null
Python null 20000
Scala 30000 null
R null null
Swift 50000 65000 输出数据帧 name merged
Java 10000
Python 20000
Scala 30000
R null
Swift 50000
浏览 23提问于2021-09-05得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
