当从RDD转换到DataFrame时，我得到一个EOFError。是什么导致了这种情况，我如何阻止它？
当从RDD转换到DataFrame时，出现EOFError可能是由于以下原因导致的：
- 数据格式错误：EOFError通常表示在读取数据时遇到了文件结束符（EOF），这可能是因为数据文件格式不正确或损坏导致的。检查数据文件的格式是否与DataFrame期望的格式相匹配，确保数据文件没有损坏。
- 数据丢失或不完整：EOFError也可能是由于数据丢失或不完整导致的。确保RDD中的数据完整且没有丢失，可以通过检查数据源或重新加载数据来解决此问题。
- 内存不足：如果转换过程中涉及大量数据，可能会导致内存不足，从而引发EOFError。可以尝试增加可用内存或优化代码以减少内存使用量。
为阻止EOFError的发生，可以采取以下措施：
- 检查数据源：确保数据源文件的完整性和正确性，可以尝试重新下载或使用其他可靠的数据源。
- 增加内存：如果内存不足导致EOFError，可以尝试增加可用内存。可以通过增加集群的内存配置或调整相关参数来实现。
- 优化代码：检查代码中是否存在内存泄漏或不必要的数据复制操作，优化代码以减少内存使用量。
- 分批处理：如果数据量过大，可以考虑将数据分批处理，以减少内存压力。可以使用Spark的分区功能将数据划分为多个小批次进行处理。
- 使用持久化存储：如果数据需要频繁转换，可以考虑将数据持久化到磁盘或其他持久化存储介质，以减少内存使用量。
请注意，以上建议是基于一般情况下的推荐做法，具体解决方法可能因实际情况而异。对于具体的代码和环境，可能需要进一步调查和分析才能找到准确的解决方案。
相关·内容
当尝试将RDD转换为DataFrame时，我得到一个错误，"EOFError“。我能做些什么来阻止这一切？ 我尝试过用另一种方法创建DataFrame，但这有其自身的复杂性。我认为我目前尝试创建DataFrame的方法是最简单的方法。 data = data.zip(bool_converted).map(lambda x: (x[0][1], x[0][2], x[0][3], x[1][1], x[0][5], x[0][6], x[0][7], x[0][8], x[0][9], x[0][10], x[0][11]))
data = data.toDF() 实际的错误消息是
浏览 31提问于2019-10-04得票数 1
回答已采纳
所以，我对火花和PySpark很陌生。我试图运行Python脚本从MySql数据库读取数据，如下代码所示：
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext
sc = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config("spark.some.config.option", "some-value") \
.get
浏览 0提问于2018-10-14得票数 0
回答已采纳
在PySpark(或Spark)中，我们可以同时使用RDD和DataFrame结构。我应该选择其中一个吗？ 我觉得有些工作用RDD很容易，有些工作用DataFrame很容易。典型的编程约定是什么？我想我用了太多的.rdd和.toDF()。
浏览 19提问于2020-09-22得票数 0
回答已采纳
目前，我可以通过RDD使用以下代码将文本文件解析为DataFrame：
def row_parse_function(raw_string_input):
# Do parse logic...
return pyspark.sql.Row(...)
raw_rdd = spark_context.textFile(full_source_path)
# Convert RDD of strings to RDD of pyspark.sql.Row
row_rdd = raw_rdd.map(row_parse_function).filter(bool)
# Conver
浏览 6提问于2016-09-02得票数 0
回答已采纳
在我们的代码中，Dataframe创建为：
DataFrame DF = hiveContext.sql("select * from table_instance");
当我将dataframe转换为rdd并尝试将其分区数作为
RDD<Row> newRDD = Df.rdd();
System.out.println(newRDD.getNumPartitions());
它将分区数减少到1(1在控制台中打印)。最初，我的dataframe有102个分区。
更新：
在阅读时，我重新划分了dataframe：
DataFrame DF = hiveContext.
浏览 2提问于2016-12-23得票数 3
我正在尝试从RDBMS创建自定义流接收器。
val dataDStream = ssc.receiverStream(new inputReceiver ())
dataDStream.foreachRDD((rdd:RDD[String],time:Time)=> {
val newdata=rdd.flatMap(x=>x.split(","))
newdata.foreach(println) // *******This line has problem, newdata has no records
})
ssc.start()
浏览 1提问于2017-02-14得票数 0
我有一个dataFrame = [CUSTOMER_ID ,itemType, eventTimeStamp, valueType]，我通过执行以下操作将其转换为RDD[(String, (String, String, Map[String, Int]))]：
val tempFile = result.map( {
r => {
val customerId = r.getAs[String]( "CUSTOMER_ID" )
val itemType = r.getAs[String]( "itemType&#
浏览 0提问于2016-10-31得票数 1
1回答
我正在尝试使用以下命令获取数据帧的分区数量：
df.rdd.getNumPartitions.toString
但当我监视spark日志时，我发现它会旋转许多阶段，这是一项代价高昂的操作。
据我所知，dataframe通过元数据向rdd添加了一个结构层。那么，为什么在转换到rdd时剥离它会花费这么多时间呢？
浏览 25提问于2020-05-20得票数 3
回答已采纳
1回答
我一直在尝试将RDDs转换为DataFrames格式，然后再转换回来。首先，我有一个类型为(Int，Int)的RDD，名为dataPair。然后，我使用以下命令创建了一个带有列标题的DataFrame对象：
val dataFrame = dataPair.toDF(header(0), header(1))
然后，我使用以下命令将其从DataFrame转换回RDD：
val testRDD = dataFrame.rdd
它返回org.apache.spark.sql.Row类型的RDD (不是(整数，整数))。然后我想用.toDF把它转换回RDD，但是我得到了一个错误：
error: va
浏览 1提问于2016-05-04得票数 11
回答已采纳
4回答
我有一个包含10个字段的数据集。我需要在这些DataFrame上执行RDD操作。是否可以执行map、flatMap等RDD操作。
下面是我的示例代码：
df.select("COUNTY","VEHICLES").show();
这是我的dataframe，我需要将这个dataframe转换成RDD，并在这个新的RDD上操作一些RDD操作。
下面是我如何将dataframe转换为RDD的代码
RDD<Row> java = df.select("COUNTY","VEHICLES").rdd();
转换为RDD后，我
浏览 2提问于2016-12-14得票数 0
1回答
我正在尝试将值数组作为新列添加到DataFrame中。
例:假设有一个Array(4,5,10)和一个dataframe
+----------+-----+
| name | age |
+----------+-----+
| John | 32 |
| Elizabeth| 28 |
| Eric | 41 |
+----------+-----+
我的要求是将上面的数组作为新列添加到dataframe中。我的预期产出如下：
+----------+-----+------+
| name | age | rank |
+----------+---
浏览 6提问于2022-01-11得票数 1
回答已采纳
我有一个包含以下示例条目的spark数据帧。
当我尝试计算数据帧中的行数时，我得到了一个Keyerror。有人能告诉我为什么会出现这个错误吗？
下面是展开的data bricks堆栈跟踪，它显示存在Keyerror。
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-42-c5195177
浏览 0提问于2016-06-04得票数 1
my_dataframe = sparkSession.createDataFrame( an_rdd, a_schema )
my_dataframe.show() # Works.
my_dataframe = my_dataframe.persist()
my_dataframe.show() # errors out.
ValueError: Length of object (69) does not match with length of fields (70) 这非常令人担忧。 进一步： my_dataframe = sparkSession.createDataFram
浏览 40提问于2020-10-14得票数 0
3回答
我正在以火花DataFrame的形式读取CSV，并在其上执行机器学习操作。我一直得到一个Python序列化EOFError --知道为什么吗？我认为这可能是内存问题--即文件超出了可用内存--但大幅减少DataFrame的大小并没有防止EOF错误。
下面是玩具代码和错误。
#set spark context
conf = SparkConf().setMaster("local").setAppName("MyApp")
sc = SparkContext(conf = conf)
sqlContext = SQLContext(sc)
#read in 5
浏览 5提问于2016-04-12得票数 34
这是我的数据
底层的RDD有两个分区。
当我做df.count时，生成的DAG是
当我做df.rdd.count时，生成的DAG是：
Ques：Count是spark中的一个操作，官方定义为“返回DataFrame中的行数”。现在，当我对dataframe执行计数时，为什么会发生洗牌？此外，当我在底层的RDD上做同样的操作时，不会发生洗牌。
--我不明白为什么会发生洗牌。，我试着看了count这里的的源代码，但对我来说没有充分的意义。被提供给该行动的“群”是罪魁祸首吗？
PS。df.coalesce(1).count不会引起任何洗牌
浏览 6提问于2017-11-09得票数 16
3回答
展示了如何使用Scala case类来推断模式，如何从RDD创建DataFrame。我试图用sqlContext.createDataFrame(RDD, CaseClass)重现这个概念，但是我的DataFrame结果是空的。下面是我的Scala代码：
// sc is the SparkContext, while sqlContext is the SQLContext.
// Define the case class and raw data
case class Dog(name: String)
val data = Array(
Dog("Rex"),
浏览 0提问于2016-05-03得票数 12
回答已采纳
我感到困惑的是，为什么在将得到的rdd.mapPartitions转换为DataFrame时，Spark似乎使用了1任务。
这对我来说是个问题，因为我想从以下几个方面着手：
DataFrame -> RDD --> rdd.mapPartitions --> DataFrame
这样，我就可以读取数据( DataFrame )，将非SQL函数应用于数据块(mapPartitions on RDD)，然后转换回DataFrame，以便我可以使用DataFrame.write进程。
我可以从DataFrame -> mapPartitions开始，然后使用像saveAsTe
浏览 4提问于2016-11-22得票数 17
回答已采纳
我不知道这问题是否较早前提出。可能是一个可能的复制，但我无法找到坚持这样做的用例。
如我们所知，我们可以直接将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
回答已采纳
我已经和Scala 2.12一起使用过Spark 3.1.2了。我想在节点之间并行一些键，它们根据接收到的键来读取数据。因此，我必须先使用RDD，然后再将其转换为Spark DataFrame。我从Oracle Database中的表中读取数据。代码如下： object managementData extends App {
val num_node = 2
def read_data(group_id: Int):String = {
val table_name = "table"
val col_name = &#
浏览 6提问于2021-09-26得票数 1
在Java中，有几种与Apache (版本1.5.2)一起使用logistic回归的选择：
spark.ml：
1) LogisticRegression lr = new LogisticRegression();
a) lr.train(dataFrame);
b) lr.fit(dataFrame);
spark.mllib：
2) LogisticRegressionWithSGD lr = new LogisticRegressionWithSGD();
a) lr.train(rdd);
b) lr.run(rdd);
3) LogisticRegressionWithLBFGS
浏览 3提问于2016-03-05得票数 2
回答已采纳
我在PySpark中这样做过：
使用DataFrame语句创建一个SELECT，以获取按资产序列号排序的资产数据，然后按时间排序。
使用DataFrame.map()将DataFrame转换为RDD。
使用整理每个资产的所有数据，使用资产的序列号作为密钥。
问题：，我能确定每个资产的数据仍然会在最后一步的RDD中按时间顺序排序吗？
时间顺序对我来说是至关重要的(我需要在每个资产的移动时间窗口上计算统计数据)。当RDD.combineByKey()为给定的密钥组合来自火花集群中不同节点的数据时，该键的数据中是否保留了任何顺序？或者，对于给定的密钥，来自不同节点的数据是否按特定
浏览 1提问于2017-04-26得票数 0
回答已采纳
\我正在尝试使用Apache Spark和Elephas并行训练多个Keras模型。下面是我要做的代码： train_data = pd.read_csv("csv_files/stats.csv")
timesteps = 30
model_1, rdd1 = train_LSTM_model(spark_context = sc, dataframe= train_data, column_number=1 ,timesteps = 30)
model_2, rdd2 = train_LSTM_model(spark_context = sc, dataframe =
浏览 29提问于2019-10-05得票数 0
我使用pyspark来运行这类连接：
rdd1=sc.textFile(hdfs_dir1).map(lambda row: (getKey1(row),getData1(row)))
rdd2=sc.textFile(hdfs_dir2).map(lambda row: (getKey2(row),getData2(row)))
result=rdd1.join(rdd2).collect()
作业执行第一个300任务的速度相当快(每个任务都有秒)，并且在到达任务301/308时挂起，即使我让它运行了几天。
我尝试使用不同的配置(工作人员数量、内存、cpus、内核、洗牌率)运行pyspark
浏览 0提问于2016-02-25得票数 4
回答已采纳
1回答
我正在尝试加载一个用spark sql以ORC格式创建的托管配置单元表。
SparkConf conf = new SparkConf().setAppName(ConnectionTest.class.getName()).setMaster(master);
JavaSparkContext context = new JavaSparkContext(conf);
SQLContext sqlContext = new HiveContext(context);
sqlContext.sql("SELECT * FROM schema.tableName").show
浏览 21提问于2017-04-20得票数 3
我以DataFrame array<string>的身份处理事务
transactions: org.apache.spark.sql.DataFrame = [collect_set(b): array<string>]
我想将其更改为RDD[Array[string]]，但是当我将其更改为RDD时，它被更改为org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]
val sam: RDD[Array[String]] = transactions.rdd
<console>:42: error: type m
浏览 14提问于2017-01-11得票数 1
来自星火的源代码：
/**
* Represents the content of the Dataset as an `RDD` of `T`.
*
* @group basic
* @since 1.6.0
*/
lazy val rdd: RDD[T] = {
val objectType = exprEnc.deserializer.dataType
rddQueryExecution.toRdd.mapPartitions { rows =>
rows.map(_.get(0, objectType).asInsta
浏览 0提问于2019-01-19得票数 11
2回答
我有一个如下形式的RDD：
org.apache.spark.rdd.RDD[(String, Int, Array[String])]
这是RDD的第一个元素：
(001, 5, Array(a, b, c))
我想将该列表分成几列，因为它是用逗号分隔的，因此预期输出将是：
(001, 5, a, b, c)
有什么帮助吗？
解决方案：
我最终解决了这个问题:我所做的是用mkstring("，")将数组组合成一个完整的字符串，然后将rdd转换为dataframe。这样，我就可以使用withColumns方法将字符串拆分成列
浏览 0提问于2016-10-03得票数 1
2回答
我有一个关于熊猫DataFrame和pd.read_pickle()函数的快速问题。基本上，我有一个大但简单的Dataframe (333mb)。当我在数据帧上运行pd.read_pickle时，我得到的是EOFError。
有什么办法可以解决这个问题吗？你知道这可能是什么原因吗？
谢谢!
浏览 3提问于2017-01-30得票数 5
如何处理时间戳列中的空值？
我将源数据保存在列表中(sql的结果)
List([222,1,222,222,2012-01-28 23:37:06.0,()],
[220,1,220,220,2012-04-24 23:37:08.0,()],
[220,1,220,220,2008-03-18 15:06:09.0,()],
...
在最后一列中，空值与时间戳混合。
但是当我试图将它作为RDD加载时
//Create RDD
val rdd = spark.sparkContext.makeRDD(res.toList)
// Create schema
浏览 0提问于2018-06-21得票数 0
回答已采纳
我是斯卡拉的新手。我读了一段无法理解的代码。有人能帮我理解下面的代码吗？
def intersectByKey[K: ClassTag, V: ClassTag](rdd1: RDD[(K, V)], rdd2: RDD[(K, V)]): RDD[(K, V)] = {
rdd1.cogroup(rdd2).flatMapValues{
case (Nil, _) => None
case (_, Nil) => None
case (x, y) => x++y
}
}
线下的意思是什么？它将如何评估？
case (
浏览 7提问于2016-03-11得票数 4
回答已采纳
我试着创建Apache Spark dataframe
val valuesCol = Seq(("Male","2019-09-06"),("Female","2019-09-06"),("Male","2019-09-07"))
valuesCol: Seq[(String, String)] = List((Male,2019-09-06), (Female,2019-09-06), (Male,2019-09-07))
模式
val someSchema = List(StructField
浏览 8提问于2019-09-19得票数 0
回答已采纳
我有两个DataFrame(Spark)，如下所示：
df1 is Array((1,WrappedArray(1,2,3)),(2,WrappedArray(1,2,4)))
org.apache.spark.sql.DataFrame = [id: int, features: array<float>]
df2 is Array((1,WrappedArray(4,5,6)),(2,WrappedArray(3,5,6)))
首先，我必须生成上述两个df的数组值的笛卡尔和求和。
例如，笛卡尔如下所示：
((11,(Array(1,2,3),Array(4,5,6))),(1
浏览 1提问于2017-11-08得票数 0
回答已采纳
我试图理解coalesce()和repartition()之间的区别。
如果我正确地理解了，coalesce()只能减少数据帧的分区数，如果我们试图增加分区的数量，那么分区的数量将保持不变。
但是当我试图执行下面的代码时，我观察到了两件事
For Dataframe与合并的分区数可以增加，如果= false，则分区数不能随着合并而增加。
是否意味着通过合并可以增加数据分区？
聚结技术在数据处理中的应用
当我执行以下代码时：
val h1b1Df = spark.read.csv("/FileStore/tables/h1b_data.csv")
println("Ori
浏览 4提问于2019-09-27得票数 3
我是通过从RDD中的文本文件加载数据来创建PySpark的。现在，我想将这个RDD转换成一个dataframe，但是我不知道RDD中有多少列和列。我正在尝试使用createDataFrame()和语法，因为它是sqlDataFrame = sqlContext.createDataFrame(rdd, schema)。我试图了解如何创建schema，但大多数示例都显示了一个硬编码模式创建示例。现在，由于我不知道列是什么，所以如何将rdd转换为dataframe？到目前为止，我的代码如下：
from pyspark import SQLContext
sqlContext = SQLContex
浏览 9提问于2017-04-07得票数 2
2回答
使用PySpark 1.6.3，我试图将RDD转换为Dataframe。这是在齐柏林飞艇笔记本上运行的测试代码。感兴趣的是rdd_ret。
>>> from pyspark.sql import Row
>>> rdd_ret.count()
9301
>>> rddofrows = rdd_ret.map(lambda x: Row(**x))
>>> things = rddofrows.take(10000)
>>> len(things)
9301
>>> [type(x) fo
浏览 0提问于2018-06-21得票数 2
好的，让我首先描述一下我是如何创建Dataframe的，以及其中包含了什么内容。
我有一组gziped HTML文档和一组指向这些HTML文档的gziped元数据
对于这两种情况，我提供了一个指向RDDs的路径列表，如下所示：
Wet_Paths_RDD = sc.parallelize(Wet_Paths)
Wet_RDD = Wet_Paths_RDD.map(open_wet_filelist).flatMap(split_wetfiles)
我以如下所示的方式准备两个RDDs：
(k,(some,other,values))
然后，我将元数据RDD与content RDD连接在一起，如
浏览 19提问于2017-01-12得票数 2
回答已采纳
我刚刚意识到我可以在Scala中做下面的事情
val df = spark.read.csv("test.csv")
val df1=df.map(x=>x(0).asInstanceOf[String].toLowerCase)
但是，在Python中，如果我试图在DataFrame上调用map函数，就会引发错误。
df = spark.read.csv("Downloads/test1.csv")
df.map(lambda x: x[1].lower())
误差
Traceback (most recent call last):
File &#
浏览 0提问于2019-09-23得票数 0
我是斯派克的新手，学习了Dataframe，操作和体系结构。在阅读RDD和Dataframe的比较时，我混淆了RDD和Dataframe的数据结构。以下是我的观察，如果错误，请帮助澄清/改正。
1)如果源数据是集群(例如: HDFS)，则RDD以分布式方式(块)存储在计算机RAM中。
如果数据源只是一个CSV文件，数据将被分发到运行服务器的RAM中的多个块(如果是膝上型计算机)。我说的对吗？
(2)分块与分块之间有何关系？哪一个是超级集？
3) Dataframe : Dataframe是否也以与RDD相同的方式存储？如果我将源数据单独存储到dataframe中，那么是否将在后台创建RDD？
浏览 0提问于2019-09-12得票数 2
回答已采纳
我想知道，与完整的RDD/DF的运行时相比，Spark的运行时是什么？我不知道这是否有区别，但我目前使用的是Java +Spark1.5.1+Hadoop2.6。
JavaRDD<Row> rdd = sc.textFile(HdfsDirectoryPath()).map(new Function<String, Row>() {
@Override
public Row call(String line) throws Exception {
String[] fields = line.split(usedS
浏览 1提问于2015-11-06得票数 0
回答已采纳
我试图从原始的Dataframe "User_df“中获取每个列的列名和数据类型，到它们自己的列中，其中列的值分别是列名和数据类型。
我有88列和9500行。
当我尝试用一个列("ColumnDataTypes")创建一个Dataframe时，下面的代码是有效的，但当我试图为这两个列创建一个Dataframe时，代码就不起作用了：
col_types = User_df.dtypes
col_types2, col_types3 = [items[:] for items in zip(*col_types)]
rdd_col_names = sc.parallel
浏览 36提问于2022-08-16得票数 0
回答已采纳
1回答
我正在研究使用ftplib (可能还有)来进行一些自动的FTP文件同步。目前，我有几台服务器可以对此进行测试，但是，当我成功地与这两台服务器进行对话时，我得到了EOFError的每一个错误回复。例如:如果我尝试使用不正确的用户/pass登录，我将得到530响应的所有内容.但是我也会得到一个EOFError；如果我用正确的用户/pass登录，或者在这样做之后尝试dir()，那么我就没有EOFError。
它似乎只出现在错误信息中。我强烈怀疑这可能是服务器造成的，而不是python:我在其他地方没有发现任何提到这个问题的地方。然而，我对服务器设置几乎没有控制。
我在问你的想法：
你知道是什么导
浏览 1提问于2011-07-25得票数 5
回答已采纳
我已经从NOSQL数据库创建了RDD，我想将RDD转换为数据帧。我已经尝试了许多选项，但都会导致错误。
val df = sc.couchbaseQuery(test).map(_.value).collect().foreach(println)
{"accountStatus":"AccountOpen","custId":"140034"}
{"accountStatus":"AccountOpen","custId":"140385"}
{"
浏览 1提问于2016-11-28得票数 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
回答已采纳
我有几个Excel文件，我需要加载和预处理，然后才能将它们加载到Spark中。我有需要处理的这些文件的列表。我这样做是为了读到：
file_list_rdd = sc.emptyRDD()
for file_path in file_list:
current_file_rdd = sc.binaryFiles(file_path)
print(current_file_rdd.count())
file_list_rdd = file_list_rdd.union(current_file_rdd)
然后，我有了一些映射函数，它将file_list_rdd从一组(路
浏览 3提问于2017-04-17得票数 0
回答已采纳
1回答
我在Spark上运行了一个迭代应用程序，我将其简化为以下代码：
var anRDD: org.apache.spark.rdd.RDD[Int] = sc.parallelize((0 to 1000))
var c: Long = Int.MaxValue
var iteration: Int = 0
while (c > 0) {
iteration += 1
// Manipulate the RDD and cache the new RDD
anRDD = anRDD.zipWithIndex.filter(t => t._2 % 2 == 1)
浏览 1提问于2019-07-30得票数 3
我有一个RDD，它是我使用map从DataFrame转过来的：
case class Record(id_1: Int, fnam_1: String, lnam_1: String, id_2: Long, fnam_2: String, lnam_2: String)
val rdd = df.map {
case Row(id_1: Int, fnam_1: String, lnam_1: String, id_2: Long, fnam_2: String, lnam_2: String) =>
Record(id_1, fnam_1, lnam_1, id_2, f
浏览 3提问于2017-04-07得票数 0
回答已采纳
我正在读一本关于Apache的书，在比较RDD和DataFrame时，它有以下几点：
RDD和DataFrame的关键区别在于，DataFrame存储的数据信息(如数据类型和列名)比RDD多得多。这允许DataFrame比在RDD上执行处理的星火转换和火花操作更有效地优化处理。
但是，在使用Scala处理RDD时，我注意到数据类型实际上是存储的。例如：
val acTuplesByAmount = acBalTuples.map{case (amount, accno) => (amount.toDouble, accno)}
acTuplesByAmount.collect()
浏览 0提问于2018-08-08得票数 0
1回答
在Spark1.6.0中，我有一个包含职务描述的列的数据框架，如：
Description
bartender
bartender
employee
taxi-driver
...
我使用以下方法从该列检索唯一值列表：
val jobs = people.select("Description").distinct().rdd.map(r => r(0).asInstanceOf[String]).repartition(4)
然后，对于每个职务描述，我尝试检索具有该职务的人员并做一些事情，但我得到了一个NullPointerException：
jobs.foreach
浏览 2提问于2016-02-21得票数 5
回答已采纳
1回答
我正在收集我的数据从卡桑德拉使用火花和火花卡桑德拉连接器。我有一个单一的共享SparkContext和REST的web应用程序。处理有下一个流程：
读卡桑德拉表
为过滤做准备(星火变换的顺序)
根据api调用参数准备RDD过滤器
在上面的算法中，对于每个调用，只有第三步是不同的(取决于api请求参数)。Api请求以并行方式执行(每个请求都是线程)。由于表中的数据不是非常动态的，而且我的火花工作人员有足够的内存来存储整个表，所以我希望在第二步之后保持我的RDD，并且对每个请求只对已经保存的RDD进行筛选。另外，我想定期更新这个RDD。实现这一目标的最佳途径是什么？
浏览 0提问于2016-07-05得票数 0
回答已采纳
我的Spark Streaming作业需要处理RDD[String]，其中字符串对应于csv文件的一行。我事先不知道模式，所以需要从RDD推断模式，然后将其内容写入parquet文件。如果我从磁盘中读取一个csv文件，我只需使用模式推断将所有内容加载到DataFrame中，并立即将其写入parquet。不过，在我的场景中，我的起点是一个RDD[String]，它是流的结果。
浏览 5提问于2017-06-14得票数 2
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
