有没有办法在mapPartitions()中返回部分结果？目前我是这样使用它的： iter: iterator[InputType] => { results.iterator}
当然，如果分区太大，results数组将抛出一个面向对象的异常所以我<
浏览 3提问于2017-01-17得票数 0
回答已采纳
我有一个作业需要在分区的spark数据帧上运行，该进程如下所示： rdd = sp_df.repartition(n_partitions, partition_key).rdd.mapPartitions(lambda x: some_function(x)) 结果是pandas.dataframe的rdd， type(rdd) => pyspark.rdd.PipelinedRDD
type(rdd.collect()[0]) => pandas.core.frame.DataFrame
浏览 124提问于2019-12-10得票数 4
回答已采纳
1回答
我在一项令我惊讶的工作中遇到了一个错误： bigger than spark.driver.maxResultSize(1024.0 MB)def add(a,b): return a+brdd有大约500个分区，func接受该分区中的行，
浏览 1提问于2017-07-26得票数 3
回答已采纳
这两者之间的区别是什么？RDD的map和mapPartitions方法？并且确实flatMap表现得像map或者像mapPartitions什么？谢谢。(编辑)即(在语义上或在执行方面)之间的区别是什么 def map[A, B](rdd: RDD[A], fn: (A => B)) rdd.mapPartitions({ iter: I
浏览 88提问于2014-01-17得票数 154
回答已采纳
我通过处理每个分区来处理流(将事件保存到HBase)，然后将每个RDD中的最后一个事件从驱动程序压缩到接收方，这样接收器就可以依次将其压缩到源。:158) at org.apache.spark.rdd.RDD.mapPartitions)
at org.apache.spark.api.java.JavaRDD.mapParti
浏览 1提问于2015-06-10得票数 1
回答已采纳
我感到困惑的是，为什么在将得到的rdd.mapPartitions转换为DataFrame时，Spark似乎使用了1任务。我可以从DataFrame -> mapPartitions开始，然后使用像saveAsTextFile这样的RDD编写器，但这并不理想，因为DataFrame.write进程可以用Orc格式覆盖和保存数据下面的工作与预期一样，为mapPartitions工作提供100个任务：
from pyspark.sql import SparkSess
浏览 4提问于2016-11-22得票数 17
回答已采纳
//work split
spark.parallelize(1, 10k).map(item => doTask(i)).collect() 其中我有一些日志记录在数据库中为doTask(i)。序列化dbManager并发送到worker节点是不容易的。有没有办法让spark指示驱动程序节点taskI已经完成，然后登录到驱动程序节点？
浏览 20提问于2021-03-20得票数 1
不错的散文，但它真正意味着什么呢？这里是一个人为的琐碎例子，不管我是否将true或false传递给mapPartitions，对于新的RDD，每个分区的数据分区仍然是相同的，即使我改变了K，V的K值，那么，意义是什么呢？一定是我错过了一些基本的东西。import org.apache.spark.HashPartitionerdef myfunc(iter:
浏览 0提问于2020-01-02得票数 2
回答已采纳
我是spark的新手，对迭代器的spark内存使用有疑问。当使用数据集的Foreach()或MapPartitions() (或者甚至直接调用RDD的迭代器()函数)时，spark是否需要首先将整个分区加载到内存中(假设分区在磁盘中)，或者当我们继续迭代时可以延迟加载数据(这意味着spark只能加载分区数据的一部分，执行任务并将中间结果保存到磁盘)
浏览 3提问于2019-04-26得票数 1
我想直接从spark worker读取Рadoop中的一些数据： 所以，在spark程序中，我有一个hadoop配置：val configuration = session.sparkContext.hadoopConfiguration，但我不能在worker中使用它，因为它不是Serializable spark.sparkContext.parallelize(paths).mapPartitions(paths => {ParquetMetadataCon
浏览 7提问于2019-10-09得票数 1
我有一个pyspark，它可以收集成一个元组列表，如下所示： (("good", "spark"), 7), (("good&
浏览 2提问于2020-06-17得票数 2
回答已采纳
目前，我正在调查我的Apache Spark集群中工作负载不平衡的可能性。是否可以在Apache Spark UI或历史服务器中查看每台机器分配的分区数量？
浏览 25提问于2019-06-28得票数 0
我试图向MQTT代理发送一个带有计算结果的DStream，但是foreachRDD一直在崩溃。 File "/
浏览 0提问于2019-07-15得票数 1
回答已采纳
我是Apache Spark的初学者。Spark的RDD API提供了像map、mapPartitions这样的转换函数。我可以理解，mapPartitions适用于RDD中的每个元素，但mapPartitions适用于每个分区，许多人都提到过，在我们想要创建/实例化对象的地方，map是理想的用法，并提供了如下示例：
valrddData = sc.textFile(&q
浏览 2提问于2018-02-26得票数 3
回答已采纳
当使用RDDs时，使用mapPartitions很好。在本例中，当使用rdd.mapPartitions(mapper).collect()时，所有工作都按预期进行。下面是下一个简单示例的输出。当只有两个分区时，我们可以3次读取函数的执行方式。其中一个分区[Row(id=1), Row(id=2)]被处理了两次。令人感到奇怪的是，其中一次执行被忽略了，正如我们在DataDrame结果中所看到的那样。(0)
rdd = spark</e
浏览 3提问于2020-10-02得票数 1
回答已采纳
1回答
有没有办法在两个并发的Spark作业中使用相同的RDD？val sc = new SparkContext(conf)
val b = a.mapPartitions(it => { Thread.sleep(5000)
浏览 4提问于2017-02-11得票数 2
1回答
我正试着给卡夫卡送一辆大的CSV。基本结构是读取CSV的一行并使用标头将其压缩。producer = SimpleProducer(kafka)使用PYSPARK，我轻松地从CSV文件创建了一个消息的RDDsentRDD = messageRDD.map(lambda x: kafkasend(x))开始搅动和发送消息
不幸的是这是非常缓慢的。这是在一个由4个c
浏览 4提问于2015-08-31得票数 6
回答已采纳
1回答
在Spark上执行FileSystem操作时，是否有任何方法从火花执行器中获取mapPartitions？如果没有，那么至少有什么方法可以获得Hadoop配置来生成一个新的Hadoop FileSystem呢？spark.sql("SELECT * FROM cities").mapPartitions{ iter =>
iter.groupedBy(some-variable
浏览 2提问于2018-06-09得票数 3
回答已采纳
我有一个用例，我需要在不中断流作业的情况下更改JSON的模式。我正在使用一个conf文件，其中我提到了所有需要的模式。我已经尝试了缓存和广播变量，通过使用单独的流水线进行持久化和非持久化，但仍然没有成功。提前感谢您的帮助！
浏览 15提问于2021-08-30得票数 0
1回答
我意识到UDF有时会减慢代码的速度。我想知道你的经历。您应用了什么UDF功能(不能仅用Pyspark代码实现)。有什么有用的UDF函数可以帮助您清理数据吗？
浏览 8提问于2022-03-03得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
