在spark中，对dataframe进行缓存操作是否会导致序列化结果异常过大？
在Spark中，对DataFrame进行缓存操作不会导致序列化结果异常过大。Spark的DataFrame是一种分布式数据集，它以逻辑和物理计划的形式表示，并且可以在内存中进行缓存以提高查询性能。
当对DataFrame进行缓存操作时，Spark会将DataFrame的数据以列存储的方式进行序列化，并将序列化后的数据存储在内存中。这种列存储的方式可以提高内存利用率和查询性能，因为它只序列化和存储每列的数据，而不是整个DataFrame的数据。
由于Spark使用了列存储和压缩等技术，所以对DataFrame进行缓存操作不会导致序列化结果异常过大。Spark会根据数据的特点进行压缩，减少存储空间的占用，并且在查询时只反序列化需要的列，从而减少了数据的传输量和内存的占用。
总结起来，对DataFrame进行缓存操作可以提高查询性能，并且不会导致序列化结果异常过大。如果需要在腾讯云上使用Spark进行数据处理和分析，可以使用腾讯云的云数据仓库CDW产品，它提供了高性能的Spark计算引擎和大规模数据存储，可以满足各种数据处理和分析的需求。更多关于腾讯云云数据仓库CDW的信息可以参考：https://cloud.tencent.com/product/cdw
相关·内容
在Spark中，对象和变量是如何保存在内存中和不同执行程序之间的？
我正在使用：
火花3.0.0
斯卡拉2.12
我正在用一个自定义的流源编写一个Spark结构化的流作业。在执行spark查询之前，我创建了一组元数据，这些元数据是由我的星火流作业使用的
我试图了解这些元数据是如何跨不同的执行器保存在内存中的？
示例代码：
case class JobConfig(fieldName: String, displayName: String, castTo: String)
val jobConfigs:List[JobConfig] = build(); //build t
浏览 3提问于2020-11-29得票数 0
回答已采纳
我有多个镶木面板文件(大约1000个)。我需要加载它们中的每一个，处理它并将结果保存到Hive表中。我有一个for循环，但它似乎只能处理2或5个文件，但不能处理1000个文件，因为似乎Spark试图同时加载它们，我需要在同一个Spark会话中单独执行。 我尝试使用for循环，然后使用for each，并使用了unpersist()，但都失败了。 val ids = get_files_IDs()
ids.foreach(id => {
println("Starting file " + id)
var df = load_file(id)
var values_df
浏览 11提问于2019-02-10得票数 1
回答已采纳
3回答
我写了下面的简单火花程序，使用火花的StreamingContext和SQLContext。
注意:这个问题是可以复制的，即使没有streamingContext。更新程序以反映同样的.
注:将spark版本降级到1.4.1 (我当时使用的是1.5.2)似乎解决了这个问题。用spark也是1.5.1这个问题我们是可以复制的.
def main(args: Array[String]) {
val sc = new SparkContext("local[*]", "test")
val sqc = new SQLContext(sc)
浏览 6提问于2015-11-25得票数 2
回答已采纳
1回答
我是新来的火种，所以希望有人能帮忙。我试图读取存储在GCP桶上的拼板文件。该文件按日期进行分区，因此，例如bucket-name/year={}/month={}/day={}
对于给定的文件，我们有以下模式描述：
直到3月份，我们以前在浮动数据类型中使用x和y列。
3月份以来，这2列现在都是双数据类型的。
从我所看到的来看，吡火花在评估浮点数方面没有任何问题，而双数据类型是兼容的数据类型。(我在网上发现的类似的错误示例与数据类型不兼容有关，例如字符串和浮点数)，但是，如果我们试图读取该文件的所有可用数据，就会遇到这个奇怪的问题：
#i.e. read all the data
浏览 4提问于2021-06-09得票数 0
我有一个500K行的spark DataFrame，它位于拼图文件中。我使用的是spark 2.0.0和Spark (RStudio和R 3.3.1)中的SparkR包，它们都运行在具有4核和8 8gb内存的本地机器上。
为了便于构建我可以在R中处理的数据集，我使用collect()方法将spark DataFrame引入R中。这样做需要大约3分钟，这比使用data.table包读取相同大小的CSV文件要长得多。
诚然，拼图文件是压缩的，解压所需的时间可能是问题的一部分，但我在互联网上找到了关于collect方法特别慢的其他评论，并且几乎没有解释的方式。
我在sparklyr中尝试过同样的操作
浏览 9提问于2016-09-19得票数 8
1回答
序列化RDD
、、
我有一个RDD，我正在尝试序列化，然后通过反序列化进行重构。我正在试着看看这在Apache中是否可行。
static JavaSparkContext sc = new JavaSparkContext(conf);
static SerializerInstance si = SparkEnv.get().closureSerializer().newInstance();
static ClassTag<JavaRDD<String>> tag = scala.reflect.ClassTag$.MODULE$.apply(Java
浏览 3提问于2015-04-10得票数 4
回答已采纳
1回答
我的用例：
从cassandra表创建数据帧。
通过对列进行过滤并修改该列的值来创建输出数据。
使用TTL集将输出数据写入cassandra，以便在短时间(2s)后删除所有修改的记录。
将输出数据返回给一个调用方，该调用者在一段时间后将其写入文件系统。我只能返回一个数据文件给调用者，我没有进一步的控制。而且，我不能增加TTL。
在执行步骤4时，输出数据为空。这是因为，星火重新评估行动的数据，由于血统，卡桑德拉查询再次完成，这现在不产生任何记录。
为了避免这种情况，我在步骤2之后添加了一个步骤：
2a) outputDataframe.cache()
这确保了在第5步中，
浏览 0提问于2018-10-17得票数 2
回答已采纳
我有一个火花-1.5.1的Hadoop2.6运行在我的本地机器上的独立模式。我正在尝试从一个示例java应用程序中运行一个hive查询，将spark.master指向在我的本地计算机上运行的火花主程序( spark ://Dynamic-i0248u:7077)。下面是一段java代码：
SparkConf sparkconf = new SparkConf().set("spark.master", "spark://impetus-i0248u:7077").set("spark.app.name", "sparkhivesqlte
浏览 2提问于2015-12-10得票数 1
2回答
星星之火SQL性能
、、、、
我的代码的算法如下
Step1.获取一个hbase实体数据到hBaseRDD
JavaPairRDD<ImmutableBytesWritable, Result> hBaseRDD =
jsc.newAPIHadoopRDD(hbase_conf, TableInputFormat.class,
ImmutableBytesWritable.class, Result.class);
Step2.将hBaseRDD转换为rowPairRDD
// in the rowPairRDD t
浏览 6提问于2014-12-25得票数 7
回答已采纳
1回答
我们正尝试使用Spark在Scala中实现 (您不需要知道回答这个问题的算法)。
该算法的项目集计算函数为freq()。代码是正确的，但是在while中的freq()函数中的每一次迭代之后，它会变慢，直到用几秒钟的时间在带有1行的表上执行交叉连接。
import System.{exit, nanoTime}
import scala.collection.mutable.WrappedArray
import org.apache.spark.sql.{Column, SparkSession, DataFrame}
import org.apache.spark.sql.functions
浏览 1提问于2020-11-14得票数 4
回答已采纳
1回答
我有两个关于火花流的问题：
我有一个按20 seconds批处理间隔运行和收集数据的火花流应用程序，在4000 batches中有由于异常而失败的18 batches：
无法计算拆分，未找到块输入-0-1464774108087。
我当时假设数据大小大于spark可用内存，而应用程序StorageLevel也是MEMORY_ONLY。
请建议如何解决这个问题。
同样在下面使用的命令中，我使用了executor内存20g(数据节点上的总RAM为140 g)，这是否意味着所有的内存都为这个应用程序保留了全部内存，如果我有多个火花流应用程序会发生什么？
在几个应用程
浏览 1提问于2016-06-01得票数 1
如果我不缓存使用带有limit选项的spark SQL生成的dataframe，当我编辑得到的dataframe并显示它时，我会得到不稳定的结果吗？ 描述。 我有一个类似下面的表格，它是通过使用带有limit选项的spark SQL生成的： +---------+---+---+---+---+
|partition| | 0| 1| 2|
+---------+---+---+---+---+
| 0| 0| 0| 10| 18|
| 1| 0| 0| 10| 17|
| 2| 0| 0| 13| 17|
+---------
浏览 6提问于2021-04-21得票数 0
回答已采纳
我正在使用withColumn和窗口操作(使用Python/Spark)构建许多新列。这导致了一个大的血统，这减慢了操作。实际上，每隔几步将DataFrame持久化到磁盘非常有帮助。
但是，如何方便地将DataFrame缓存在内存中并强制物化，从而减少沿袭呢？
(我见过关于count的建议，但我的印象是只适用于RDDs?)
浏览 0提问于2017-02-14得票数 2
回答已采纳
1回答
我有两个关于星火串行化的问题，我可以简单地通过谷歌找不到答案。
如何打印当前使用的序列化程序的名称；我想知道spark.serializer是Java还是Kryo。
我有下面的代码，它应该使用Kryo序列化；用于dataframe的内存大小变为21 same，这是在没有序列化的情况下缓存时的四分之一；但是当我删除Kryo配置时，大小保持相同的21 same。这是否意味着Kryo从一开始就没有被使用过？可能是因为dataframe中的记录只是行，Java和Kryo序列化都是相同的大小吗？
val conf =新的SparkSession.builder.master("local
浏览 1提问于2017-12-26得票数 5
当我尝试缓存()或持久化(MEMORY_ONLY_SER())我的RDD时，火花集群挂起。它工作得很好，计算结果在7分钟左右。如果我不使用cache()。
我有6个c3.x大型EC2实例(4个核，每个7.5GBRAM)，总共提供了24个核心和37.7GB。
我在master上使用以下命令运行我的应用程序：
./uber-offline.jar:/root/spark/assembly/target/scala-2.10/spark-assembly_2.10-0.9.0-incubating-hadoop1.0.4.jar SPARK_MEM=5g MEMORY_FRACTION="0
浏览 2提问于2014-03-29得票数 3
回答已采纳
在Spark中对DataFrame进行排序时，幕后会发生什么？
例如,
df = spark.read.format('csv').option('foo')
df.sort(i)
我知道当您将数据读取到DataFrame时会发生什么，但是我很好奇当您排序时会发生什么？与MapReduce相比有什么不同？
浏览 2提问于2018-08-14得票数 0
我正在使用以下环境：spark = 2.0.0、hdp = 2.5.3.0、python = 2.7、yarn-client
我的PySpark代码大部分时间都工作得很好。
然而，有时我在df.count()函数上遇到以下异常
适用于我的代码：
df= spark.read.orc("${path}")
df.count()
我得到异常的代码：
df= spark.read.orc("${path}")
df = df.cache()
df.count()
堆栈跟踪：
Job aborted due to stage failure: Task 0 in s
浏览 0提问于2017-01-19得票数 0
我想知道如果我的应用程序使用DataFrame API，什么时候使用spark.serializer？
如果我理解得好的话：
不用于缓存或用于闭包序列化和广播变量的shuffling.Only的。
对吗？
浏览 39提问于2018-06-08得票数 0
1回答
我从parquet中读取了一个DataFrame，并希望在选择一些嵌套结构后对其进行缓存。
df.select($"a.b.c" as "c").cache()
我知道整个a列将从输入中读取(Spark2.5。应该解决这个问题：)，但我想知道存储是否会更聪明，只存储选择的结果(所以不是整个a)。
浏览 0提问于2018-10-02得票数 0
1回答
我正在尝试写数据帧到MySql DB和使用Apache Spark 2.3.1。它有20K到30K的行从mySql读取，并使用20个分区进行分区。我首先过滤数据帧，并尝试将过滤后的结果集写入mysql DB。 但是写操作变得太慢。在没有过滤的情况下，df写操作正在按照预期的速度和性能执行。有人能帮上忙吗？ 我的代码： dataFrame = spark.read.format('jdbc').option(...).load()
//performing some operations and adding new column "total" in data
浏览 24提问于2021-10-15得票数 0
我想不断地详细说明dataset流的行(最初由Kafka启动)：基于一个条件，我想更新Radis哈希。这是我的代码片段(lastContacts是前一个命令的结果，该命令是这种类型的流：org.apache.spark.sql.DataFrame = [serialNumber: string, lastModified: long] )。这将扩展到org.apache.spark.sql.Dataset[org.apache.spark.sql.Row])：
class MyStreamProcessor extends ForeachWriter[Row] {
override def
浏览 0提问于2019-03-26得票数 0
回答已采纳
不明白为什么缓存的DFs (特别是第一个)在Spark中根据代码段显示不同的Storage Levels
print(spark.version)
2.4.3
# id 3 => using default storage level for df (memory_and_disk) and unsure why storage level is not serialized since i am using pyspark
df = spark.range(10)
print(type(df))
df.cache().count()
print(df.storageLeve
浏览 2提问于2019-07-07得票数 1
我有一个DataFrame，我将它映射到RDD ()中来测试SVMModel。
我正在使用齐柏林飞艇和星火1.6.1
这是我的代码：
val loadedSVMModel = SVMModel.load(sc, pathToSvmModel)
// Clear the default threshold.
loadedSVMModel.clearThreshold()
// Compute raw scores on the test set.
val scoreAndLabels = df.select($"features", $"label")
浏览 12提问于2016-05-13得票数 2
回答已采纳
运行以下代码：
(1)
val resultDf = myDataFrame.map(row => { println(s"$row"); return row })
我可以看到每一行的打印结果(使用“纱线日志-applicationId xxxx”获取日志)。但是，当处理代码比较复杂时：
(2)
val resultDf = myDataFrame.map(row => { println(s"$row"); /* complex processing code */})
我发现打印出的数量是实际行数的2到3倍。但在这两种情况下myDataFram
浏览 2提问于2022-04-23得票数 0
1回答
我还有一项任务要做。
用户在执行submit命令时提供一组配置文件的IP地址。
假设该数组如下所示：
val ips = Array(1,2,3,4,5)
数组中最多可以有100.000个值。
对于数组中的所有元素，我应该读取Cassandra的数据，执行一些计算并将数据插入Cassandra。
如果我这样做了：
ips.foreach(ip =>{
- read data from Casandra for specific "ip" // for each IP there is different amount of data to read (within the
浏览 4提问于2016-04-21得票数 0
2回答
根据
星火中的所有转换都是懒惰的，因为它们不计算结果--正确的away...This设计使Spark能够更有效地运行。
有时，我需要对我的数据文件立即执行某些操作。但是，由于dataframe是“延迟计算的”(如上文所述)，所以当我在代码中编写这些操作时，很难保证Spark实际上将按照代码的其余部分执行这些操作。例如：
val someDataFrame : DataFrame = getSomehow()
val someOtherDataFrame : DataFrame = getSomehowAlso()
// Do some stuff with 'someDataF
浏览 3提问于2016-09-08得票数 10
回答已采纳
2回答
我是星火框架的新手，在我的本地机器上做一些小任务来练习。我的任务是:我在S3中存储了365个压缩的csv文件，其中包含每天的日志。我想要建一个全年的数据集。我的方法是从桶中检索密钥，构建每日数据格式，将它们统一为月份数据，对它们进行同样的操作，并作为回报获得全年数据。
它适用于我检索到的用于测试的一些样本数据。在构建DataFrames之前，我对文件进行解压缩，将未压缩的csv文件写入磁盘，并使用它创建DataFrame。
问题是:如果我从磁盘中删除csv文件(使其成为临时文件)，在创建dataframe之后，我无法对dataframe执行任何操作(例如year_df.count())。抛出S
浏览 0提问于2017-01-09得票数 1
回答已采纳
我对火花的缓存机制有点困惑。
假设我有一个Spark应用程序，在多个转换结束时只有一个操作。在其中，假设我有一个dataframe A，并在其上应用了2-3转换，创建了多个数据文件，这最终有助于创建最后一个要保存到磁盘中的数据。
例子：
val A=spark.read() // large size
val B=A.map()
val C=A.map()
.
.
.
val D=B.join(C)
D.save()
那么，为了提高性能，我需要缓存dataframe吗？
提前谢谢。
浏览 0提问于2019-12-04得票数 6
回答已采纳
我是Scala/Spark的新手，希望你们能帮助我。我想在一个hdfs的目录中获取在某个时间戳之后创建的文件，以便在Zeppelin中进行一点监控。因此，我需要一个包含文件名、文件大小和modificationDate的列。 我发现这对我来说很有效，可以获得我需要的所有信息： val fs = FileSystem.get(new Configuration())
val dir: String = "some/hdfs/path"
val input_files = fs.listStatus(new Path(dir)).filter(_.getModificationT
浏览 18提问于2020-12-10得票数 1
回答已采纳
我正在Azure DBR7.3LTS，SMAR3.0.1，Scala2.12上运行以下代码，这些代码位于Standard_E4as_v4 (32.0 GB内存，4核，1 DBU)和驱动程序(56.0GB内存，16核，3 DBU)的集群上
目的是处理5.5TB的数据
我面临以下例外："org.apache.spark.SparkException:由于阶段失败而中止的作业: 57071个任务中的1163个任务(4.0 GiB)的序列化结果的总大小大于spark.driver.maxResultSize 4.0 GiB“，在6.1min内，有148.4 GiB正在处理数据。
我不收集或传输
浏览 3提问于2020-11-02得票数 0
回答已采纳
我正在开发一个Spark程序，并收到以下异常：
16/11/07 15:58:25 ERROR yarn.ApplicationMaster: User class threw exception: java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds]
java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds]
at scala.concurrent.impl.Promise$Default
浏览 5提问于2016-11-07得票数 22
回答已采纳
我想知道，与完整的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
回答已采纳
我正在尝试编写一个基于Spark的GraphFrames的scala类。GraphFrame类被定义为here。GraphFrames类扩展了serializable。我想写一个类来扩展GraphFrames并计算一些额外的图形属性。为了简化这个例子，我创建了一个不包含任何函数的类。它所做的只是扩展GraphFrames： import org.apache.spark.sql.DataFrame
import org.graphframes._
class NewGraphFrame(@transient private val _vertices: DataFrame,
浏览 25提问于2021-01-06得票数 3
回答已采纳
1回答
最近，我开发了一个使用Scala和Spark的Spark流应用程序。在这个应用程序中，我广泛地使用了隐式类(Pimp模式)来实现更通用的实用程序，比如通过创建一个扩展Spark的Dataframe的隐式类来将Dataframe写入HBase。例如,
implicit class DataFrameExtension(private val dataFrame: DataFrame) extends Serializable { ..... // Custom methods to perform some computations }
但是，我的团队的一位高级架构师重构了代码(指定了一些风格不
浏览 3提问于2018-09-10得票数 0
回答已采纳
有时(例如，为了测试和标记)，我想强制执行在DataFrame上定义的转换。AFAIK调用像count这样的操作并不能确保所有的Columns都是实际计算的，show可能只计算所有Rows的一个子集(参见下面的示例) 我的解决方案是使用df.write.saveAsTable将DataFrame写到HDFS，但是这会“扰乱”我的系统，我不想再保存更多的表。 那么触发DataFrame求值的最佳方式是什么呢 编辑： 请注意，在spark开发人员列表上还有一个最近的讨论：http://apache-spark-developers-list.1001551.n3.nabble.com/Will-
浏览 62提问于2017-03-10得票数 20
回答已采纳
1回答
Spark任务不可序列化
、、、
卡桑德拉和DataStax社区，我有一个问题，我希望有一个明智的人可以帮助我。
我们正在迁移我们的变异代码从Hadoop到星火运行在卡桑德拉之上(通过DataStax企业)。DSE 4.7在生产中，4.8在发展中。
在生产中使用Java 7，在开发中使用Java 7/8。
我们需要几个DataFrame转换，我们认为通过Spark对内存中的DataFrame DataFrame编写一个UDF将完成这项工作。其中最主要的是：
我们的数据的每一个文本值都是前缀和后置“。例如，“一些数据”，这是非常恼人的，所以我们想要清理每一个这些。
我们希望添加一个列，该列包含从许多其他列组成的散列键。
浏览 3提问于2016-03-23得票数 1
回答已采纳
2回答
在星火外壳上，我使用下面的代码从csv文件中读取
val df = spark.read.format("org.apache.spark.csv").option("header", "true").option("mode", "DROPMALFORMED").csv("/opt/person.csv") //spark here is the spark session
df.show()
假设这将显示10行。如果我通过编辑在csv中添加一个新行，那么调用df.show()是否会再次显示新行？
浏览 5提问于2016-12-05得票数 1
回答已采纳
1回答
在我的Scala/火花应用程序中，我创建了DataFrame。我计划在整个程序中多次使用这个Dataframe。这就是为什么我决定使用.cache()方法来实现DataFrame。正如您在循环中看到的那样，我使用不同的值对DataFrame进行了多次筛选。由于某种原因，.count()方法将返回始终相同的结果。实际上，它必须返回两个不同的计数值。此外，我注意到Mesos中的奇怪行为。感觉好像没有执行.cache()方法。在创建DataFrame之后，程序将进入代码if (!df.head(1).isEmpty)的这一部分并执行很长一段时间。我假设缓存过程将运行很长时间，而其他进程将使用此缓存并
浏览 2提问于2020-01-27得票数 0
回答已采纳
来自诸如和 one这样的问题，我问自己，当我将一个在RDD级别上分区的数据文件保存到一个(例如)拼花表中时，spark.rdd.compress是否也有效果。
或者换句话说：spark.rdd.compress是否也压缩了我在使用dataframe.write.saveAsTable(...)时创建的表
从中获取，spark.rdd.compress执行以下操作：
是否压缩序列化的RDD分区(例如，Java中的StorageLevel.MEMORY_ONLY_SER或Python中的StorageLevel.MEMORY_ONLY )。可以节省大量的空间，而代价是额外的CPU时间。压缩将使用
浏览 0提问于2019-06-06得票数 0
回答已采纳
1回答
在对原始数据集/dataframe执行几个转换(联合)后，我希望将dataset/dataframe写入csv。获得的dataset/dataframe显示得很完美，没有任何问题，但是当我想在csv中编写dataset/dataframe时，它会显示以下错误
Caused by: java.io.FileNotFoundException: File file:/../file.csv does not exist
It is possible the underlying files have been updated. You can explicitly invalidate the c
浏览 17提问于2022-06-13得票数 1
当我在EMR上运行我的spark作业(版本2.1.1)时，每次运行都会计算数据帧上不同数量的行数。我首先从s3读取数据到4个不同的数据帧，这些计数总是一致的，然后在连接数据帧后，连接的结果具有不同的计数。之后，我还过滤了结果，并且在每次运行时也有不同的计数。变化很小，1-5行的差异，但这仍然是我想要理解的东西。
这是用于连接的代码：
val impJoinKey = Seq("iid", "globalVisitorKey", "date")
val impressionsJoined: DataFrame = impressionDsNoDu
浏览 0提问于2017-10-22得票数 1
我有一个不可序列化的对象，我需要这个对象一直100%地缓存在内存中。然而，由于spark job由于负载增加而需要更多的执行内存，它试图将缓存的对象逐出到磁盘中，并且由于对象不可序列化，我将得到"task not serializable“异常。有没有办法强制spark将我的对象一直保存在缓存中？
浏览 6提问于2017-02-22得票数 0
基本上，我想检查日期数据中的每个值是否都是正确的格式"MM/dd/yy“。
val df: DataFrame = spark.read.csv("----")
但是，每当我应用函数映射时：
df.map(x => right_format(x)).show()
试着展示这个新的数据格式/数据集，我得到了一个不可序列化的错误。
有人知道为什么吗？
我尝试过使用intellij调试器进行调试，但没有效果。
val df: DataFrame = spark.read.csv("----")
df.map(x => right_format(x)
浏览 0提问于2019-07-10得票数 0
回答已采纳
如何在spark scala流应用程序中清理JVM占用的内存。我正在运行60秒时间间隔的流式作业。在我的前六个小时，没有问题，在那之后，我面对的是JVM堆内存问题。有没有办法在spark scala中以编程方式清理我的GC或JVM内存？
在我的应用程序中，我使用了Dataframe，registertemptable也结束了我的程序，我将结果写入到HDFS中。目前在我的应用程序spark SQL上下文级，我正在取消缓存，像这样我们有没有其他方法可以释放内存？
错误消息:异常在线程"dag-scheduler-event-loop“java.lang.outofmemoryError:
浏览 1提问于2017-09-16得票数 2
我正在做一个测试数据接近100万条记录和4个这样的文件的项目。任务是执行大约40次计算，连接来自4个不同文件的数据，每个文件接近1gb。
目前，我使用可保存的将每个数据保存到spark表中，并执行操作。例如- table1与table2联接，并将结果保存到table3。Table3(1和2的结果)与table4等连接。最后，我将这些计算保存在不同的表中，并生成报告。
整个过程大约需要20分钟，我担心的是，当这段代码得到的数据可能是该代码的5倍时，会出现性能问题。
还是最好以分区的方式从每个文件中保存这些数据，然后执行联接并到达最终的结果集。
P.S --目标是获得即时的结果，在某些情况下，用户
浏览 0提问于2018-08-17得票数 0
回答已采纳
我的Spark脚本遇到了一个小问题。基本上，我有原始数据，在分组和计数之后进行聚合，等等，我希望将输出保存为特定的JSON格式。
编辑：
我试图简化这个问题，并改写了它：
当我从源dataframe中选择列名有别名的Array[org.apache.spark.sql.Column]数据时，在试图将行映射到case类时使用列名(实际上是索引)作为变量，那么我就会得到一个“任务不可序列化”的异常。
var dm = sqlContext.createDataFrame(Seq((1,"James"),(2,"Anna"))).toDF("id",
浏览 1提问于2016-12-20得票数 3
我有一个非常简单的类，我正在尝试使用spark来减少它。由于某些原因，它不断抛出异常，不能序列化类。这是我的班级：
@SerialVersionUID(1000L)
class TimeRange(val start: Long, val end: Long) extends Serializable {
def this(){
this(0,0)
}
def mergeOverlapping(rangesSet : Set[TimeRange]) = {
def minMax(t1: TimeRange, t2: TimeRange) : TimeRange
浏览 0提问于2016-04-28得票数 1
1回答
在下面的示例中，我误解了PySpark的性能。
我有几个DataFrame，因此我加入了它们。
print"users_data"
print users_data.show()
print"calc"
print calc.show()
print"users_cat_data"
print users_cat_data.show()
data1 = calc.join(users_data, ['category_pk','item_pk'], 'leftouter')
print "
浏览 0提问于2018-01-15得票数 2
回答已采纳
我感到困惑的是，为什么在将得到的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
回答已采纳
我有一个数据设置与火花-卡桑德拉-连接器1.6.2.我试着用卡桑德拉进行一些转换。Datastax企业版为5.0.5。
DataFrame df1 = sparkContext
.read().format("org.apache.spark.sql.cassandra")
.options(readOptions).load()
.where("field2 ='XX'")
.limit(limitVal)
.reparti
浏览 1提问于2017-08-09得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
