我是Python spark的新手，我在Jupyter笔记本中运行以下spark代码并获取AttributeError: 'NoneType' object has no attribute '_jvm_9, 3)parallel.mapPartitions(f).collect() 运行代码时，可以在下面找到完整的错误ipython-input-55-44576a0dc413> in <modu
浏览 34提问于2021-02-08得票数 0
回答已采纳
我试图在PySpark上运行PySpark，但我得到了一个ImportError。有没有人知道我该怎么解决这个问题？谢谢你的帮助！！(lambda行:iterateRDD(行，模型))文件"C:\Python\Python27\lib\site-packages\pyspark
浏览 2提问于2019-07-17得票数 0
1回答
我得到了一段模拟的数据，就像：import pysparksqlContext = SQLContext(sc)
columnMock我正在考虑使用python中的map类型进行翻译，这样我就可以使用</e
浏览 0提问于2018-06-02得票数 0
回答已采纳
我是Pyspark的新手，我正在试着做一个简单的计数。然而，它给了我这个错误。文本文件在hdfs中。/rdd.py", line 1008, in count File"/usr/local/lib/spark-2.0.1-bin-hadoop2.7/python/pyspark&
浏览 0提问于2016-10-24得票数 7
1回答
您应用了什么UDF功能(不能仅用Pyspark代码实现)。有什么有用的UDF函数可以帮助您清理数据吗？除了Pyspark文档之外，还有什么资源可以帮助我学习UDF函数吗？
浏览 8提问于2022-03-03得票数 0
当使用RDDs时，使用mapPartitions很好。在本例中，当使用rdd.mapPartitions(mapper).collect()时，所有工作都按预期进行。但是，当转换为DataFrame时，一个分区将被处理两次。下面是下一个简单示例的输出。当只有两个分区时，我们可以3次读取函数的执行方式。令人感到奇怪的是，其中一次执行被忽略了，正如我们在</
浏览 3提问于2020-10-02得票数 1
回答已采纳
1回答
在转换中，用户应该知道，如果任务或作业阶段被重新执行，每个任务的更新可能不止一次应用。
rdd.mapPartitions(accum_func).mapPartitions(lambda
浏览 3提问于2017-03-29得票数 0
回答已采纳
在Monterey v12.6.1上本地运行下面的错误。添加了更多的上下文rejected_on_load_df = enriched_df.rdd.mapPartitions(process_rows).toDF(
enriched_df.schema.add(&
浏览 5提问于2022-11-28得票数 0
在尝试用MQTT发布我的结果之前，我尝试了saveAsFiles()，这起作用了(但不是我想要的)。/pyspark/rdd.py", line 806, in foreachPartition
self.mapPartitions(func).count() # Force evaluationFile "/SPARK_HOME/python/lib/pyspark.zip/pyspark/rd
浏览 0提问于2019-07-15得票数 1
回答已采纳
基本上，我在这段代码中出错了：其中data是RDD，我的helper定义为： c = locations[x](位置只是一个数据点的数组)我看不出问题是什么，但我也不是pyspark的佼佼者，所以有人能告诉我为什么我从这段代码中得到的'PipelinedRDD‘对象是不可迭代的吗？
浏览 0提问于2016-04-19得票数 6
回答已采纳
1回答
我尝试使用sql上下文创建jsonrdd，但是得到了下面的错误。我把代码和错误放在了下面。请建议..。sqlContext.jsonRDD('/idn/home/ijamalmo/ibfiles/datafiles/sampledata.json')文件""，第1行，在文件"/opt/mapr/spark/spark-1.4.1/python/pys
浏览 1提问于2016-08-23得票数 0
2回答
有没有办法在Spark中获取map任务的ID？例如，如果每个map任务调用一个用户定义的函数，我能否从该用户定义的函数中获取该map任务的ID？
浏览 0提问于2015-07-26得票数 12
因此，我正在尝试使用Python (Pyspark)来学习火花。我想知道函数mapPartitions是如何工作的。这就是它所需要的输入和它提供的输出。我在网上找不到合适的例子。[ [1, 2, 3], [3, 2, 4], [5, 2, 7] ]
我想从所有的列表中删除元素2，如何使用mapPartitions实现这一点。
浏览 1提问于2014-11-04得票数 33
回答已采纳
我在Google Cloud Dataproc上使用Spark streaming来执行一个框架(用Python编写)，它由几个连续的管道组成，每个管道代表Dataproc上的一个作业，它基本上是从Kafka/pyspark/rdd.py", line 747, in foreach
self.mapPartitions(processPartition).count() # Force evaluationFile "/usr/lib/spa
浏览 0提问于2016-05-24得票数 3
我感到困惑的是，为什么在将得到的rdd.mapPartitions转换为DataFrame时，Spark似乎使用了1任务。)，将非SQL函数应用于数据块(mapPartitions on RDD)，然后转换回DataFrame，以便我可以使用DataFrame.write进程。我可以从DataFrame -> mapPartitions开始，然后使用像saveAsTextFile这样的RDD编写器，但这并不理想，因为DataFrame.write
浏览 4提问于2016-11-22得票数 17
回答已采纳
2回答
下面是可再现性的代码(将在PySpark中运行)。这一问题突出显示在使用与vs没有火花的烦恼时所见的不同。
浏览 7提问于2016-02-03得票数 7
回答已采纳
1回答
在了解了Apache文档之后，我尝试使用mapPartition模块进行实验。在下面的代码中，我希望看到初始的RDD，就像在函数myfunc中一样，我只是在打印值之后返回迭代器。但是，当我在RDD上执行collect时，它是空的。from pyspark import SparkConf n = 5
rdd = sc.parallelize(
浏览 0提问于2017-03-23得票数 1
回答已采纳
我正在尝试修改Pyspark源代码，以便模拟一个不确定的场景，以便测试一些配置更改。具体地说，我将rdd.py更改为休眠一段时间，以模拟工作节点停止，例如保存到s3。当我在Pyspark shell中运行代码时，我可以观察到更改的效果，但当我运行spark-submit时，我无法观察到更改的效果。即使我在本地运行，也会发生这种情况。我不明白在shell中调用Pyspark的方式与通过spark-submit调用<em
浏览 1提问于2018-02-01得票数 0
我想使用PySpark在每个节点上执行一些内容，如下所示： def f (i): vv = rdd.mapPartitions(f).collect()
但我不明白为什么我要为此加载一个文件
浏览 0提问于2016-05-23得票数 0
回答已采纳
我在我的s3存储桶中有一个数据集，"sqlnew" test是目录，test是我的文件，当我试图执行下面的pyspark代码抛出错误时。vfttt+A9yqtt+114AttttttttvKejCevccc")执行count后抛出错误"/home/hadoop/spark/s
浏览 0提问于2018-04-04得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
