1回答
我有一个包含名为primary_use的列的pyspark DataFrame。 下面是第一行： ? 为了创建一个布尔向量来指示某一行中的primary_use是Education还是Office，我使用了以下代码。这会导致异常： def is_included_in(row):
building.foreachTraceback (mos
浏览 129提问于2019-12-13得票数 1
4回答
如何在Python Spark structured streaming中使用foreach在输出上触发操作。query = wordCounts\ .outputMode('update')\ .start()
def
浏览 0提问于2018-01-11得票数 2
我在pyspark 2.3.4中使用结构化流。我试图按以下方式使用foreach操作符：我得到以下错误：
这里有什么问题？？
浏览 1提问于2019-11-09得票数 1
我试图在PySpark上运行PySpark，但我得到了一个ImportError。有没有人知道我该怎么解决这个问题？谢谢你的帮助！！(lambda行:iterateRDD(行，模型))文件"C:\Python\Python27\lib\site-packages\pyspark\rdd
浏览 2提问于2019-07-17得票数 0
我试图从hdfs中获取数据，并对每个数据进行迭代，以便对列_c1进行分析。import findsparkimport pysparkscnamenode:9000/data.csv', header=False, inferSchema= True)df.forEach
浏览 3提问于2022-04-07得票数 0
回答已采纳
1回答
我正在使用PySpark在木星上的Azure。我试图使用UDF在dataframe上进行测试，但是UDF没有执行。在下一个单元格中，我尝试执行一个简单的udf。def iterateMeals(user):这不会产生任何输出。我希望数据文件中的每一项都会被打印出来。我还试过使用pyspark.sql.functions
from
浏览 0提问于2016-03-24得票数 1
回答已采纳
当从终端运行以下简单脚本时，它在pyspark中工作得很好：foo = sc.parallelize([1,2])但是当在Rodeo中运行时，它会产生一个错误，其中最重要的一行是：
Exception: Python in worker has different version2.7 than that in driver 3.5, PySpark<
浏览 2提问于2016-04-13得票数 0
回答已采纳
我需要使用Pyspark从Elasticsearch读取数据。我正在尝试在pyspark中设计一个流，如下所示-ii) foreach -in-rdd1"org.elasticsearch.hadoop.mr.LinkedMapWritablesc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat"，\ "org.apa
浏览 0提问于2016-07-11得票数 0
1回答
我在脚本中使用for循环为size_DF(数据帧)的每个元素调用一个函数，但这需要很长时间。我尝试通过逐个映射删除for循环，但是我没有得到任何输出。size_DF是我从表中获取的大约300个元素的列表。print "length: ", length使用地图newObject.full_item(sc, dataBase, len(l[0]), end_dat
浏览 0提问于2018-02-28得票数 2
回答已采纳
有没有办法利用map函数将pyspark dataframe的每一行存储到自定义的python类对象中？pyspark dataframe 例如，在上面的图片中，我有一个spark数据帧，我想将id、features、label的每一行都存储到一个节点对象中(具有3个属性node_id、node_features我想知道这在pyspark中是否可行。我试过像这样的东西 对于df.rdd.collect() do_something ( row )中的行 但这不能处理大数
浏览 19提问于2020-07-14得票数 0
Spark 1.6.1，pyspark的问题{"event":4,"Userid":12345,"time":123456789,"device_model":"iPhone阅读spark streaming编程指南中的“使用foreachRDD的设计模式”一节
在数据集之外做一些事情时，似乎建议使用foreachRDD。在我的例子中，我希望通过网络将数据写入HBase，因此我对我的流数据<
浏览 2提问于2016-05-28得票数 1
谁知道我是否可以在Databricks的SQL中进行迭代构造(如loop，while，for)？
浏览 35提问于2021-11-11得票数 0
我遇到了Spark2.3的pyspark.ml.image新特性的问题。
AttributeError：'NoneType‘对象没有属性'_jvm’。您可以在pyspark中尝试以下代码，并在文件夹"jpg“中使用已准备好的图片。例如，我将放在其中。在“本地计算
浏览 0提问于2018-06-22得票数 1
我是pySpark的新手。我收到了一个csv文件，大约有1000列。我正在使用databricks。大多数这些列之间有空格，例如“总收入”，“总年龄”等。我需要更新所有的列名与空格下划线‘_’。我已经试过了但它在databricks上的Pyspark中不起作用。
浏览 1提问于2019-08-02得票数 1
我想用Python (PySpark)实现从Kafka源代码到MariaDB (PySpark)的Spark结构化流(Spark2.4.x)。from pyspark.sql import SparkSessionfrom pyspa
浏览 3提问于2019-11-08得票数 7
回答已采纳
2回答
我正在尝试使用pyspark来解决以下问题。我有一个文件在hdfs上的格式，这是一个转储的查找表。key2, value2 (k,v) = line.split(",", 1)kvfile.
浏览 0提问于2014-07-01得票数 13
回答已采纳
1回答
我是Pyspark中的nooby，我假装玩了几个函数，以便更好地理解如何在更现实的场景中使用它们。有一段时间，我尝试对RDD中的每个数字应用一个特定的函数。我的问题基本上是，当我尝试打印从我的RDD中抓取的内容时，结果是无from pyspark import SparkConf , SparkContext
numbersRDD = sc.parallelize(n
浏览 0提问于2018-01-25得票数 3
回答已采纳
我面临以下问题：lst = []
lst = lst.append(x.firstname)
# df.foreach(lambda x: func2(x))
pr
浏览 0提问于2021-11-23得票数 0
我在Google Cloud Dataproc上使用Spark streaming来执行一个框架(用Python编写)，它由几个连续的管道组成，每个管道代表Dataproc上的一个作业，它基本上是从Kafka在Dataproc上运行这个Spark streaming框架在5月初(准确地说是5月3日)之前一直相当稳定:我们开始遇到频繁的套接字超时异常，这会终止我们的管道。它在一天中也是随机发生的，我检查了可能相关的代码更改，但我找不到任何更改。此外，这似乎只发生在有4个工作节点的集群上，而有3个节点的集群上的管道非常相似，根
浏览 0提问于2016-05-24得票数 3
云服务器
ICP备案
对象存储
云直播
实时音视频
腾讯云开发者
