我正在纠结于mapPartitions的正确用法。import spark.implicits._val newDF= dataDF.mapPartitions( iterator => {
浏览 5提问于2018-07-24得票数 0
回答已采纳
一些消息来源，比如Mathei的，提到星火DataFrames是建立在RDD之上的。我在类中发现了一些关于RDDs的地方(在Spark2.0中，我必须看一下DataSet)；但是对于这两个API是如何在幕后结合在一起的，我的理解仍然非常有限。有人能解释一下DataFrames是如何扩展RDD的吗？
浏览 3提问于2016-08-29得票数 2
回答已采纳
4回答
如果我调用map或mapPartition，并且我的函数接收来自PySpark的行，那么创建本地PySpark或Pandas DataFrame的自然方法是什么？合并行并保留架构的东西？def combine(partition): dfpart = pd.DataFrame(rows,columns=rows[0].keys())
浏览 0提问于2015-12-23得票数 2
回答已采纳
现在，我想使用Spark在分布式环境中创建myTable，就像我的数据集在HDFS中一样。我以前使用过星火，我对它很熟悉。 return table.append(record);});
如果我知道如何命令RDD取行中的某些no，那么我可以使用它以分布式方式创建
浏览 2提问于2015-06-22得票数 0
回答已采纳
我感到困惑的是，为什么在将得到的rdd.mapPartitions转换为DataFrame时，Spark似乎使用了1任务。)，将非SQL函数应用于数据块(mapPartitions on RDD)，然后转换回DataFrame，以便我可以使用DataFrame.write进程。我可以从DataFrame -> mapPartitions开始，然后使用像saveAsTextFile这样的RDD编写器，但这并不理想，因为DataFrame.write进程可以用Orc格
浏览 4提问于2016-11-22得票数 17
回答已采纳
(SparkContext.scala:1435) at org.apache.spark.api.java.JavaRDDLike$class.mapPartitions(JavaRDDLike.scala:141)
at org.apache.spark.api.java.JavaRDD.mapPartitions(JavaRDD.scala<em
浏览 1提问于2015-06-10得票数 1
回答已采纳
1回答
我使用Spark将一个非常大的TSV文件转换成两组键值对，然后加载到分布式密钥值存储中。我不使用DataFrames，因为TSV不遵循定义得很好的模式，而稀疏矩阵则是更好的模型。我的伪码大概是，val entityAttrPairs = orig.mapPartitions(convertLinesToKVPairs)
v
浏览 1提问于2016-05-18得票数 0
回答已采纳
问题概要:假设我在AWS中的EMR集群上使用spark处理了300+ GB的数据。该数据有三个属性，用于在Hive中对文件系统进行分区:日期、时间和(比方说) anotherAttr。我们使用Parquet并不是因为组织上的原因。
这种方法运行得相当好，并解决了下游团队使用Hive而不是Spark看不到由大量文件导致的性能问题的问题。例如，如果我使用整个300 GB的DataFrame，使用1000个分区(在spark中)和相关的列进行重新分区，并将其
浏览 0提问于2020-05-09得票数 0
回答已采纳
//work split
spark.parallelize(1, 10k).map(item => doTask(i)).collect() 其中我有一些日志记录在数据库中为doTask(i)。有没有办法让spark指示驱动程序节点taskI已经完成，然后登录到驱动程序节点？
浏览 20提问于2021-03-20得票数 1
这里是一个人为的琐碎例子，不管我是否将true或false传递给mapPartitions，对于新的RDD，每个分区的数据分区仍然是相同的，即使我改变了K，V的K值，那么，意义是什么呢？import org.apache.spark.HashPartitionerdef myfunc(iter: Iterator[(String,map(x => ("RFD"+x+"ABC", (1000, "xc888x"+x)))
浏览 0提问于2020-01-02得票数 2
回答已采纳
1回答
我有两个Spark DataFrames，具有匹配的分区对。我希望将每对分区发送到不同的执行器，并在它们上执行Python函数。如果我只有一个DataFrame，我可以使用mapPartitions，但是我不能访问另一个。我试过广播第二个，但它太大了，如果没有分区就无法放在我的executors上。我想我可能需要使用pyspark.rdd.RDD中的函数，但我不确定是哪一个。有几个选项，比如groupByKey和cogroup似乎是相关的，但我不明白如何将它们应用于DataFrame对象。
浏览 18提问于2019-07-04得票数 0
我在一个库中使用DataFrame mapPartitions，它是的松散实现。输出DataFrame有一些新的(大)列，输入DataFrame在执行mapPartitions之前进行了分区和内部排序。大多数用户会在其他列上进行投影，然后在已分区的列上聚合。由于mapPartitions使用planWithBarrier，这会导致昂贵的冗余混洗。我想知道在catalyst api中是否有一个不太复杂的解决方案？//key1->key2 many to one
.repar
浏览 40提问于2019-08-01得票数 0
1回答
那里。我对myself非常陌生，我自己也在学习UDF。我意识到UDF有时会减慢代码的速度。我想知道你的经历。您应用了什么UDF功能(不能仅用Pyspark代码实现)。有什么有用的UDF函数可以帮助您清理数据吗？除了Pyspark文档之外，还有什么资源可以帮助我学习UDF函数吗？
浏览 8提问于2022-03-03得票数 0
1回答
有没有办法在两个并发的Spark作业中使用相同的RDD？val sc = new SparkContext(conf)
val b = a.mapPartitions(it => { Thread.sleep(5000); it })
// Compu
浏览 4提问于2017-02-11得票数 2
1回答
基本结构是读取CSV的一行并使用标头将其压缩。a = dict(zip(header, line.split(",")message = json.dumps(a)producer = SimpleProducer(kafka)使用
浏览 4提问于2015-08-31得票数 6
回答已采纳
我有一个作业需要在分区的spark数据帧上运行，该进程如下所示： rdd = sp_df.repartition(n_partitions, partition_key).rdd.mapPartitions现在我希望将结果转换成spark数据帧，我这样做的方式是： sp = None if i == 0:sp = spark.createDataFrame(partition) sp =
浏览 124提问于2019-12-10得票数 4
回答已采纳
是否可以在Spark Structured streaming中使用mapPartitions？遇到这些错误dataframe_python.mapPartitions(processfunction)选项2：‘带有流源的查询必须使用writeStream.start()；
有没有办法
浏览 15提问于2019-08-01得票数 1
我正在使用databricks运行它，在某些时候，我看到了这样的图片：注意，在第12阶段，我只有一个分区--这意味着没有并行性。我怎么能推断出原因呢？添加(稍微模糊的)代码： .createO
浏览 1提问于2018-06-19得票数 2
回答已采纳
2回答
我正尝试在安装了Spark 1.5.1的SparkR上执行随机森林算法。我不清楚，为什么我会得到这个错误-此外，即使我在代码中使用mapPartitions函数，我也会收到错误消息-请查找以下代码：
rdd <- SparkR:::textFile(sc, "http:
浏览 0提问于2016-02-10得票数 0
1回答
我尝试使用sql上下文创建jsonrdd，但是得到了下面的错误。我把代码和错误放在了下面。请建议..。sqlContext.jsonRDD('/idn/home/ijamalmo/ibfiles/datafiles/sampledata.json')文件""，第1行，在
文件"/opt/mapr/spark/spark-1.4.1/python&#x
浏览 1提问于2016-08-23得票数 0
云服务器
ICP备案
即时通信 IM
对象存储
云直播
腾讯云开发者
