如果select条件基于RDD转换，spark会将整个表加载到内存中吗？
如果select条件基于RDD转换，Spark不会将整个表加载到内存中。RDD（弹性分布式数据集）是Spark的核心数据结构，它代表了分布在集群中的不可变分区数据集合。RDD的转换操作是惰性执行的，只有在遇到行动操作时才会触发实际的计算。
在Spark中，select操作通常是通过使用transformations（转换操作）来实现的，例如map、filter等。这些转换操作只会对RDD中的每个分区进行计算，并不需要将整个表加载到内存中。这种分区计算的方式使得Spark能够有效地处理大规模数据集，而无需将所有数据加载到内存中。
需要注意的是，如果在select操作之后执行了行动操作（如collect、count等），Spark会将所需的数据加载到内存中进行计算。但是，这仅限于所需的数据，而不是整个表。
总结起来，基于RDD转换的select操作不会将整个表加载到内存中，而是按需计算每个分区的数据。这种方式使得Spark能够高效地处理大规模数据集。
相关·内容
2回答
我们现有的应用程序在启动时将大约一千万行从数据库加载到对象集合中。该集合存储在GigaSpaces缓存中。
当应用程序接收到新消息时，将检查缓存以查看该消息的条目是否已经存在。如果不是，则根据消息中的数据将新实体添加到缓存中。(同时，新实体被持久化到数据库中)。
我们正在研究使用Spark和Scala重新设计应用程序的可行性和附加值。问题是，在Spark中建模的正确方式是什么。
我的第一个想法是从数据库加载到Spark RDD中。查找现有条目显然很简单。但是，因为RDD是不可变的，所以向缓存添加新条目需要进行转换。考虑到大量的数据集，我的假设是这将不会有很好的表现。
另一个想法是将缓存创建为一
浏览 2提问于2016-01-25得票数 2
我遇到了这个问题，一切都很好，但当我使用df.cache()时，它会导致org.apache.spark.memory.SparkOutOfMemoryError问题。 有没有人有过类似的问题？ 谢谢! 以下是导致该问题的代码行： df.cache()
df = df.select(
*df.columns,
f.greatest(*custom_columns).alias(f"custom_max"),
f.least(*custom_columns).alias(f"custom_min"),
)
return df 下面是错误日
浏览 368提问于2021-04-17得票数 0
我正在读一本书，这本书对我来说是一个矛盾的陈述。引用书中的话：“默认情况下，每次对Spark运行操作时，都会重新计算Spark的RDDs。”但在接下来的几行中，它声明：“在第一次计算之后，Spark将把RDD内容存储在内存中，并在以后的操作中重用它们。”
我的问题是，如果RDDs存储在内存中，为什么每次在它们上调用操作时都要重新计算它？
在第一个语句中，RDD每次都会重新计算，在第二个语句中，RDD被存储在内存中，以便在将来的操作中重用它们。
浏览 11提问于2021-02-27得票数 0
我正在尝试理解Apache spark中的延迟计算。我的理解是：
假设我在hardrive中有一个文本文件。
步骤：
1)首先我将创建RDD1，它现在只是一个数据定义(现在没有数据加载到内存中)。
2)我在RDD1上应用了一些转换逻辑并创建了RDD2，这里RDD2仍然是数据定义(仍然没有数据加载到内存中)
3)然后我在RDD2上应用过滤器并创建RDD3 (仍然没有数据加载到内存中，RDD3也是一个数据定义)
4)我执行了一个操作，这样我就可以在文本文件中获得RDD3输出。所以当我执行这个操作的时候，我期望从内存中输出一些东西，然后spark将数据加载到内存中，创建RDD1，2和3，并产生输出。
浏览 2提问于2018-03-04得票数 1
我有一个RDD，它是通过读取一个大小约为117 of的本地文本文件形成的。
scala> rdd
res87: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[3] at textFile at <console>:24
我缓存RDD:-
scala> rdd.persist()
res84: rdd.type = MapPartitionsRDD[3] at textFile at <console>:24
在此之后，我将RDD上的‘采取(1)’操作称为强制求值。完成后，我将检查Spark的存储页面。
浏览 4提问于2015-07-20得票数 3
我试图理解map和flatMap是如何工作的，但被下面的代码卡住了。flatMap()函数返回一个RDDChar，但我期望返回的是RDDString。有人能解释一下为什么它会产生RDDChar吗？
scala> val inputRDD = sc.parallelize(Array(Array("This is Spark"), Array("It is a processing language"),Array("Very fast"),Array("Memory operations")))
scala> val
浏览 17提问于2017-06-25得票数 1
回答已采纳
1回答
我在火花执行过程中完全混乱。我参考过可能的文章和教程，没有人在详细讨论。我可能误解了火花。请纠正我。
我的40 my文件分布在10个节点集群的4个节点(每个节点10 my)上。当我在代码中说spark.read.textFile("test.txt")时，它会将所有4个节点的数据(40 my )加载到驱动程序(主节点)中吗？否则，这个RDD将分别加载到所有4个节点中。在这种情况下，每个节点RDD应该保存10 it的物理数据，是吗？整个RDD保存10 in数据，并为每个分区执行任务，即spark 2.0中的128 in。最后，将输出移到驱动程序(主节点)。
我在某个地方读到
浏览 0提问于2019-04-03得票数 3
回答已采纳
下面是的一个示例。
lines = spark.textFile("hdfs://...")
errors = lines.filter(_.startsWith("ERROR"))
errors.persist()
论文系统：“注意到基本的RDD (行)没有加载到内存中，这是可取的，因为错误消息可能只是数据的一小部分(小到足以容纳内存)”。
我的问题是如何判断RDD是否会加载到ram中？
浏览 3提问于2015-03-05得票数 1
回答已采纳
2回答
我很难理解RDD分区和HDFS输入分块之间的区别。因此，本质上，当您提交星火申请时：
当Spark应用程序希望从HDFS读取时，HDFS上的文件将有输入分块(假设每个输入分块64 mb，每个输入分块都存在于不同的数据节点上)。
现在假设Spark应用程序希望使用(sc.textFile(PATH_IN_HDFS))从HDFS加载该文件。所述文件约为256 mb，具有4个输入分块，其中2个分块位于数据节点1上，另2个分块位于数据节点2上。
现在，当Spark将这256 mb加载到它的RDD抽象中时，它会将每个输入分块(64 Mb)加载到4个不同的RDD中(其中数据节点1中有2个RDD，数据节点1
浏览 0提问于2016-10-08得票数 2
我有一个RDD (字符串数组) org.apache.spark.rdd.RDD[String] = MappedRDD[18]，并将其转换为具有唯一Ids的映射。我做了'val vertexMAp = vertices.zipWithUniqueId‘，但这给了我另一个'org.apache.spark.rdd.RDD[(String, Long)]'类型的RDD，但我想要一个'Map[String, Long]’。如何转换我的'org.apache.spark.rdd.RDD[(String, Long)] to Map[String, Long]‘
浏览 4提问于2014-10-14得票数 5
回答已采纳
在运行spark 2.2.0文档中的示例Tfidf代码时，链接如下：
代码如下：
sentenceData = spark.createDataFrame([
(0.0, "Hi I heard about Spark"),
(0.0, "I wish Java could use case classes"),
(1.0, "Logistic regression models are neat")
], ["label", "sentence"])
tokenizer = Tokeniz
浏览 1提问于2018-07-06得票数 1
我是spark的新手，对迭代器的spark内存使用有疑问。
当使用数据集的Foreach()或MapPartitions() (或者甚至直接调用RDD的迭代器()函数)时，spark是否需要首先将整个分区加载到内存中(假设分区在磁盘中)，或者当我们继续迭代时可以延迟加载数据(这意味着spark只能加载分区数据的一部分，执行任务并将中间结果保存到磁盘)
浏览 3提问于2019-04-26得票数 1
2回答
转换基于现有的RDD创建新的RDD。基本上，RDDs是不可变的，而Spark中的所有转换都是惰性的。RDDs中的数据在执行操作之前不会被处理，但是如果不处理数据，如何创建新的RDDs？例如，在filter操作中，如何在不实际将RDD加载到内存并进行处理的情况下创建新的RDD？
浏览 2提问于2016-09-03得票数 4
回答已采纳
运行DSE 4.7
假设我有一个4节点的DSE Cassandra/星火星系团.
我有一张卡桑德拉表，里面有四百万份记录。
在Spark上运行以下Spark "select * from table where email =？或mobile = ?“
是否会将所有数据加载到RDD中，然后根据where子句进行筛选？每个火花节点是否每个节点有1,000,000条记录加载到内存中？
浏览 3提问于2015-06-11得票数 1
回答已采纳
1回答
我知道要创建一个RDD，我们有两种方法：
将驱动程序program.Referencing数据中的现有集合从诸如HDFS、HBase等外部存储系统中分离出来的Par等位化。
但是，我想知道从数据湖(如Azure DataLakeStorageGener2，ADLS Gen 2)读取数据时会发生什么。例如，如果我有以下命令：
df = spark.read.csv("path to ADLS Gen 2").rdd
我想知道这些数据是如何读取的；它是否被写入驱动程序？还是直接进入工作节点？
那么，如果我们在Dataframe或RDD上应用了一些转换，那么处理在哪里进行呢？只有将数
浏览 3提问于2019-10-16得票数 1
回答已采纳
正如问题所表明的那样，我已经尝试了相当一段时间来编写一些可以读取数据帧的代码，并将顶点添加到gremlin图中，并从dataframe提取属性。为此，我编写了以下代码：
val graph = TinkerGraph.open()
val g = graph.traversal
def myFunction(field1:String,field2:String) ={
graph.addVertex(field1,field2)
}
val df = List(
(1,"A","X",1),
(2,"B","
浏览 0提问于2018-02-14得票数 1
我不知道这问题是否较早前提出。可能是一个可能的复制，但我无法找到坚持这样做的用例。
如我们所知，我们可以直接将csv文件加载到dataframe，并可以将其加载到RDD中，然后稍后将该RDD转换为dataframe。
RDD = sc.textFile("pathlocation")
我们可以在此RDD上应用一些Map、filter和其他操作，并将其转换为数据格式。
此外，我们还可以创建直接读取csv文件的dataframe。
Dataframe = spark.read.format("csv").schema(schema).option("head
浏览 0提问于2018-11-29得票数 0
回答已采纳
1回答
我是新来的火种，所以希望有人能帮忙。我试图读取存储在GCP桶上的拼板文件。该文件按日期进行分区，因此，例如bucket-name/year={}/month={}/day={}
对于给定的文件，我们有以下模式描述：
直到3月份，我们以前在浮动数据类型中使用x和y列。
3月份以来，这2列现在都是双数据类型的。
从我所看到的来看，吡火花在评估浮点数方面没有任何问题，而双数据类型是兼容的数据类型。(我在网上发现的类似的错误示例与数据类型不兼容有关，例如字符串和浮点数)，但是，如果我们试图读取该文件的所有可用数据，就会遇到这个奇怪的问题：
#i.e. read all the data
浏览 4提问于2021-06-09得票数 0
1回答
火花数据帧
、、
星火数据帧和Pandas数据帧的区别是什么？我的理解是，Pandas数据框架主要用于将CSV数据读入DF，在DF中，Spark数据帧用于将弹性分布式数据( RDD )加载到数据帧中，并且可以操作数据帧。请分享你的反馈
浏览 6提问于2020-09-01得票数 0
回答已采纳
1回答
我使用Spark2.3.2训练了LDA(特征维度: 100，迭代: 100，分布式版本，ml )。在此之后，我希望使用该模型来转换新的DataSet。但是当我转换新的数据时，我总是会得到与错误相关的内存错误。我将数据大小从x0.1更改为x0.01。但是总是得到内存错误( java.io.ByteArrayOutputStream.hugeCapacity(ByteArrayOutputStream.java:123)中的java.lang.OutOfMemoryError)
当数组的大小超过hugeCapacity - 8时，就会发生Integer.MAX_VALUE错误(溢出)，但我将数据大
浏览 6提问于2019-03-05得票数 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回答
我正在广泛地使用Spark，Spark的核心是RDD，正如RDD论文所示，在流应用程序方面也有局限性。这是RDD文件的准确引文。
正如在介绍中所讨论的，RDDs最适合于将相同操作应用于dataset所有元素的批处理应用程序。在这些情况下，RDDs可以高效地将每个转换记为谱系图中的一个步骤，并且可以恢复丢失的分区，而不必记录大量数据。RDDs不太适合于对共享状态进行异步细粒度更新的应用程序，例如web应用程序的存储系统或增量web爬虫。
我不太明白为什么RDD不能有效地管理状态。星火流如何克服这些限制？
浏览 2提问于2016-03-06得票数 2
在foreachRDD()中，有一些使用Spark的示例。但是如果我想在tranform()中使用SQL
case class AlertMsg(host:String, count:Int, sum:Double)
val lines = ssc.socketTextStream("localhost", 8888)
lines.transform( rdd => {
if (rdd.count > 0) {
val t = sqc.jsonRDD(rdd)
t.registerTempTable("logstash")
浏览 3提问于2015-02-15得票数 1
我目前正在使用一个连接到RDS的EMR集群来收集两个表。
创建的两个RDD相当庞大，但我可以执行其他的.take(x)操作。
我还可以执行更复杂的操作，例如：
info_rdd = somerdd.map(lambda x: (x[1], x[2])).groupByKey().map(some_lambda)
apps_rdd = apps.join(info_rdd).map(lambda x: (x[0], (x[1][0], x[1][1][0], x[1][1][1])))
但是，执行以下操作来计算从RDS导入的不同用户的数量不起作用：
unique_users = rdd.dist
浏览 2提问于2017-02-14得票数 2
回答已采纳
有谁能纠正我对坚持火种的理解吗？
如果我们在RDD上执行了缓存()，那么它的值只缓存在那些最初计算RDD的节点上。这意味着，如果有一个由100个节点组成的集群，并且在第一和第二节点的分区中计算RDD。如果我们缓存了这个RDD，那么Spark将只在第一个或第二个工作节点中缓存它的值。因此，当这个Spark应用程序在以后的阶段尝试使用这个RDD时，那么Spark驱动程序必须从第一个/第二个节点获取值。
我说的对吗？
(或)
是RDD值持久化在驱动程序内存中而不是节点上吗？
浏览 3提问于2016-08-28得票数 23
回答已采纳
在Spark的介绍中，它说
。
据我所知，您必须使用.cache()或.persist().If手动缓存RDD。我两种方法都不采用，如下所示
val file = sc.textFile("hdfs://data/kv1.txt")
file.flatMap(line => line.split(" "))
file.count()
我没有将RDD“文件”保存在缓存或磁盘中，在这种情况下，Spark能比MapReduce运行得更快吗？
浏览 1提问于2014-09-10得票数 4
1回答
通过蒙特卡罗近似计算π的简单pyspark程序：
def inside(p):
x, y = p
return x*x + y*y < 1
num_samples = 10000000
count = sc.parallelize(((random.random(), random.random()) for i in range(num_samples))).filter(inside).count()
pi = 4 * count / num_samples
print(pi)
为了理解RDDs的缓存，我尝试了以下方法：
rdd = sc.parallelize((
浏览 0提问于2018-04-10得票数 1
回答已采纳
我正在写一份spark工作，根据学生日期过滤最新的学生记录。但当我尝试使用数十万条记录时，它工作得很好。但是，当我使用大量记录运行它时，我的sparkjob返回下面的错误。
我猜这个错误是因为我从表中加载了所有数据并将int放入了RDD中。因为我的表包含大约420万条记录。如果是这样的话，有没有更好的方法来有效地加载这些数据并成功地继续我的操作？
请任何人帮我解决这个问题
WARN TaskSetManager: Lost task 0.0 in stage 1.0 (TID 1, 10.10.10.10): java.lang.OutOfMemoryError: GC overhead li
浏览 22提问于2017-03-10得票数 1
1回答
我有一份很大的文件。对于文件的每一行(两个字)，我必须检查整个文件。我在scala中所做的显然是错误的，但我不知道如何修复它。
此函数返回文件的所有行(大约300万！)
def allSentences() : ArrayList[String] = {
val res: ArrayList[String] = new ArrayList[String]()
val filename = "/path/test.txt"
val fstream: FileInputStream = new FileInputStream(filename)
浏览 1提问于2017-07-09得票数 1
回答已采纳
我已经成功安装了Apache Spark，Hadoop over Ubuntu 12.04 (单机模式)用于Logistic回归。也测试了小的csv数据集，但它不能在269369行的大型数据集上工作。
library(SparkR)
sc <- sparkR.init()
iterations <- as.integer(11)
D <- 540
readPartition <- function(part){
part = strsplit(part, ",", fixed = T)
list(matrix(as.numeric(unlist(par
浏览 1提问于2014-10-07得票数 2
我有pyspark，其尺寸为(28002528,21)，并试图使用以下代码行将其转换为熊猫数据：
pd_df=spark_df.toPandas()
我发现了一个错误：
第一部分
Py4JJavaError: An error occurred while calling o170.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 39.0 failed 1 times, most recent failure: Lost task 3.0
浏览 0提问于2019-02-25得票数 5
1回答
Spark流不会持久化信息
、、、、
我已经创建了一个类似于wordcount的Spark流脚本。优点是，我希望将所有信息存储在一个集合(addedRDD)中，但过了一段时间后，由于块消失了，一个异常被启动。有没有办法将这个累积的RDD保存在内存中？
import org.apache.spark._
import org.apache.spark.streaming._
import scala.collection.mutable
import org.apache.spark.rdd.RDD
val ssc = new StreamingContext(sc, Seconds(5))
val lines = ssc.soc
浏览 2提问于2017-10-26得票数 0
我正在运行一个具有以下属性的星火作业：
"spark.driver.disk": "10g",
"spark.driver.maxResultSize": "40g",
"spark.driver.memory": "50g",
"spark.driver.memoryOverhead": "85g",
"spark.executor.cores": "5",
"spark.executor.disk&
浏览 19提问于2022-05-11得票数 0
2回答
我们都知道斯派克在内存中做计算。我只是对下面的内容感到好奇。
如果我从HDFS中在我的RDD外壳中创建了10个pySpark，这是否意味着所有这10个RDD的数据都将驻留在Spark内存中？
如果我不删除RDD，它会永远存在于记忆中吗？
如果我的数据集(文件)大小超过可用RAM大小，数据将存储在哪里？
浏览 5提问于2016-11-22得票数 21
回答已采纳
1回答
我用的是电火花
我读取了一个libsvm文件，转换它，然后再保存它。
我将每个数据行保存为具有稀疏数据的对象。
我尝试使用MLUtils.saveaslibsvm，而不是使用MLUtils.loadlibsvm读取文件，并得到以下错误
ValueError:无法将字符串转换为浮动：[
在org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193) at org.apache.spark.api.python.PythonRunner$$anon$1.(PythonRDD.scala:234) at org
浏览 2提问于2017-08-10得票数 0
回答已采纳
1回答
spark如何处理并发查询？我读过一些关于spark和底层RDD的文章，但我不能理解如何处理并发查询？
例如，如果我运行一个查询，将数据加载到内存中，并且整个可用内存被消耗，同时其他人运行一个涉及另一组数据的查询，spark如何为这两个查询分配内存？另外，如果考虑到优先级，会有什么影响。
另外，运行大量并行查询会导致机器挂起吗？
浏览 2提问于2014-11-11得票数 1
1回答
我正在使用Apache Spark运行一个独立的应用程序，当我将所有数据作为文本文件加载到RDD时，我得到了以下错误：
15/02/27 20:34:40 ERROR Utils: Uncaught exception in thread stdout writer for python
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
at java.nio.ByteBuffer.allocate(ByteBuffe
浏览 10提问于2015-02-28得票数 0
我有一个非常大的json文件集(超过一百万个文件)，我想和Spark一起工作。
但是，我以前从来没有尝试过将这么多数据加载到RDD中，所以我实际上不知道是否可以这样做，或者甚至不知道是否应该这样做。
在Spark的RDD(s)中处理这些数据量的正确模式是什么？
浏览 0提问于2016-01-13得票数 0
我正试图找出与Word2Vec一起使用哪种presented库，并根据pyspark文档向我提供了两个选项。
https://spark.apache.org/docs/2.2.0/mllib-feature-extraction.html#word2vec https://spark.apache.org/docs/2.2.0/ml-features.html#word2vec
mllib似乎用于使用RDD，而ml似乎使用"DataFrames“。
有什么关系？如果这是引擎盖下的火花，他们不是都在使用RDD吗？
这里的"DataFrame“是什么？因为文件无法解释这一点。
浏览 0提问于2018-12-12得票数 6
回答已采纳
我有一个spark应用程序，其中我需要将数据从executors获取到driver，并且我使用的是collect()。然而，我也遇到了toLocalIterator()。据我在互联网上读到的toLocalIterator()，它返回一个迭代器，而不是立即发送整个RDD，所以它有更好的内存性能，但速度呢？当涉及到执行/计算时间时，collect()和toLocalIterator()之间的性能如何？
浏览 0提问于2017-06-04得票数 7
我知道localCheckpoint删除了重建RDD所需的历史记录。cache正在保存RDD的当前状态，因此不需要重新构建它。
然而，我在几个方面感到困惑。如果我使用localCheckpoint，并且我需要在稍后的代码中使用这个RDD，我经常会得到一个关于如何找不到分区的Exception。
我查看了sparkUI中的Storage选项卡，它显示只节省了一小部分RDD，比如17%。
所以我读了更多的书，意识到spark将丢弃旧的RDDs。有没有办法让Spark永远拥有它？
另外，如果我使用cache而不是localCheckpoint，这个问题会得到解决吗？但这只需要时间，因为Spark将不
浏览 0提问于2018-10-05得票数 2
Spark提供了一种方法saveAsTextFile，可以方便地将RDD[T]文件存储到磁盘或硬盘中。
T是一个任意可序列化的类。
我想反转这个操作。我想知道有没有可以很容易地将文件加载到RDD[T]中的loadFromTextFile
让我说清楚：
class A extends Serializable {
...
}
val path:String = "hdfs..."
val d1:RDD[A] = create_A
d1.saveAsTextFile(path)
val d2:RDD[A] = a_load_function(path) // this is
浏览 1提问于2015-05-15得票数 5
回答已采纳
我正在尝试使用count()方法来计数以下RDD元素。第一项建议如下：
scala> val data_wo_header=dropheader(data)
data_wo_header: org.apache.spark.rdd.RDD[String]
当我真的指望这个的时候，我会得到：
scala> data_wo_header.count()
res1: Long = 20000263
这个操作比较快，大约需要26秒。
现在，我对RDD进行如下转换：
scala> val ratings_split = data_wo_header.map(line => li
浏览 1提问于2017-03-12得票数 0
回答已采纳
我试图从employeeId中获取employee_table，并使用这个id查询employee_address表以获取地址。
桌子没什么问题。但是当我运行下面的代码时，我得到了org.apache.spark.SparkException: Task not serializable
我想我知道这个问题。问题是，sparkContext是与主人，而不是与工人。但我不知道该怎么处理这事。
val employeeRDDRdd = sc.cassandraTable("local_keyspace", "employee_table")
try {
v
浏览 0提问于2016-12-14得票数 1
回答已采纳
我有一个流应用程序，可以获取数据流并在数据流上运行sql操作，然后将其转储到文件中
dstream.foreachRDD { rdd =>
{spark.read.json(rdd)
.select("col")
.filter("value = 1")
.write.csv("s3://..")
现在，我需要能够在我的计算中考虑到之前的计算(来自较小的批次)(类似于以下内容)：
dstream.foreachRDD { rdd =>
{val df = spark.read.json(rdd)
val prev_df =
浏览 0提问于2017-11-20得票数 0
我在scala中有以下实用程序函数：
object MyUtiltity {
def processData(data1: org.apache.spark.rdd.RDD[String], data2: org.apache.spark.rdd.RDD[String],
data3: org.apache.spark.rdd.RDD[String]) = {
function1(data1, data3)
function2(data2, data3)
}
private def function1 {...}
private def fun
浏览 0提问于2015-06-05得票数 0
回答已采纳
最近我在Spark中使用了KernelDensity类，我试着在windows10中将它序列化到我的磁盘上，下面是我的代码：
// read sample from disk
val sample = spark.read.option("inferSchema", "true").csv("D:\\sample")
val trainX = sample.select("_c1").rdd.map(r => r.getDouble(0))
val kd = new KernelDensity().setSample(tr
浏览 1提问于2020-05-10得票数 0
1回答
我有数to的数据要使用Apache Spark处理。我使用代码sparkContext.binaryFiles(folderpath)加载文件夹中的所有数据。我认为它会将全部数据加载到RDD中，并导致OutOfMemory错误。
如何将1TB数据拆分为250 it，并让RDD加载它？
浏览 1提问于2016-02-03得票数 1
我在Spark1.5.2(每20小时可复制一次)上有一个"GC开销上限“，我的代码中没有内存泄漏。会不会是斯派克的错？因为Spark1.6.0，他们改变了内存管理，它能解决这个问题吗？
2016-09-05 19:40:56,714 WARN TaskSetManager: Lost task 11.0 in stage 13155.0 (TID 47982, datanode004.current.rec.mapreduce.m1.p.fti.net): java.lang.OutOfMemoryError: GC overhead limit exceeded
at jav
浏览 2提问于2016-09-06得票数 0
回答已采纳
1回答
我有以下代码，其中错误在sc.parallelize()
val pairs = ret.cartesian(ret)
.map {
case ((k1, v1), (k2, v2)) => ((k1, k2), (v1.toList, v2.toList))
}
for (pair <- pairs) {
val test = sc.parallelize(pair._2._1.map(_._1 ))
}
哪里
k1，k2是字符串
v1，v2是双打的列表
每当我试图访问sc时，我都会得到以下错误。我在这里做错什么了？
线程
浏览 10提问于2015-09-18得票数 2
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
