在pandas_udf spark中返回一个Pandas序列

文章/答案/技术大牛

发布

2回答

使用部分函数的pyspark pandas_udf出错

python-3.x、pyspark、pandas-groupby、user-defined-functions

我定义了一个pandas udf函数，并希望将除pandas.Series或pandas.DataFrame之外的其他参数传递给udf函数。我想使用partial函数来做这件事，但它出错了。import SparkConf, SparkContextfrom pyspark.sql.functions import pandas_udf, PandasUDFType conf = SparkConf().setMaster(&q

浏览 15提问于2019-12-18得票数 0

回答已采纳

1回答

PySpark访问DataFrame列的foreachPartition()自定义函数

python、apache-spark、dataframe、iterator、pyspark

我有一个叫做“内部”的函数。我想要将这个函数应用到中。为此，我在我创建的dataframe上调用"foreachPartition(inside)“方法。"我也试着用row.lon，row.select来做，把它当作一个列表，但是无法得到所需的结果。

浏览 1提问于2018-05-22得票数 0

回答已采纳

1回答

pandas_udf给出了与pyarrow相关的错误

python、pandas、apache-spark、pyspark、pyarrow

我有一个数据帧，我想要在其中使用pysaprk中的polyline库获取给定地理位置的lat_long +-----------------+--------------------+--------|2020-03-03| 我正在使用pandas_udf，并且apache箭头已启用。from pyspark.sql.functions import col, pandas_udf spark.conf.set("spark.sql.execution.arrow.pyspar

浏览 30提问于2020-08-31得票数 0

回答已采纳

1回答

apache-spark、pyspark、apache-arrow

在Apache Spark上，我有一个应该返回pd.Series的pandas_udf函数，如何将其存档？我试过了：def udfreturn pd.Series([1,2,3,4,5])pyarrow.lib.ArrowNotImplementedE

浏览 36提问于2020-02-27得票数 0

1回答

Spark中的用户定义函数(UDF)是否在集群工作节点上并行运行？

apache-spark、user-defined-functions

假设我在python中创建了一个函数，所以将一个数字求幂为2： def squared(s): return s * s 然后我在Spark session中注册了这个函数，如下所示： spark.udf.register("squaredWithPython", squared) 然后当我在Spark SQL中调用UDF时，如下所示： spark.range(1, 20).

浏览 33提问于2019-10-10得票数 0

回答已采纳

1回答

可能将数据分解为拓扑的各个部分

python、pandas、apache-spark

我有一千万的记录数据。我的要求是，我需要对熊猫的这些数据做一些操作，而且我对所有1000万条记录都没有记忆。所以我希望能够对每个块进行分块并使用toPandas#do chunking to take X records at a timep_df = chunked_df.toPandas()如何通过记录计数将我的数据分割成相等的x-部分或部分，比如一次一百万。这两种解决方案都是

浏览 1提问于2018-10-26得票数 10

回答已采纳

1回答

如何在结构化查询中使用scikit学习模型？

apache-spark、scikit-learn、pyspark、spark-structured-streaming

我尝试过使用pandas_udf (版本1)，它给了我这个错误：代码：import pandas as pd from pyspa

浏览 4提问于2019-11-20得票数 1

回答已采纳

1回答

带有PySpark 2.4的Pandas UDF

pandas、apache-spark、pyspark、apache-spark-sql

我正在尝试基于下面的spark文档使用PySpark 2.4，pyarrow版本0.15.0和pandas版本0.24.2执行pandas_udf，在调用pandas_udf函数时有问题。import pandas as pdfrom pyspark.sql.types import LongType# Declare the function and create t

浏览 1提问于2020-11-06得票数 0

1回答

PySpark:带有标量Pandas的无效returnType

apache-spark、pyspark、apache-arrow

我正在尝试从pandas_udf返回一个特定的结构。它在一个集群上工作，但在另一个集群上失败。我尝试在组上运行udf，这需要返回类型为数据框架。from pyspark.sql.functions import pandas_udfimport numpy as np #Calculate distance, return sc

浏览 2提问于2018-03-26得票数 5

回答已采纳

1回答

在spark数据帧上实现pythonic统计函数

pandas、apache-spark、pyspark、statistical-test

我在spark数据帧中有非常大的数据集，它们分布在节点上。我可以使用spark库pyspark.sql.functions进行简单的统计，如mean、stdev、skewness、kurtosis等。但为了做到这一点，我必须将spark数据帧转换为pandas，这意味着将数据强制到主节点中，如下所示： import scipy.stats as statsJBtest=stats.jarqu

浏览 30提问于2020-09-13得票数 0

1回答

熊猫udf在火花放电中的窗口功能

python、sql、pandas、apache-spark、pyspark

目标是使用作为pyspark中的窗口函数。下面是一个很小的例子。df是一个pandas DataFrame和一个spark表：from pyspark.sql import SparkSession 1 1 2 33 2 4 75 3 6 6

浏览 3提问于2022-04-30得票数 1

回答已采纳

5回答

如何修复"ImportError: PyArrow >= 0.8.0必须安装；但是，没有找到“？

apache-spark、pyspark、pyspark-sql

我使用PySpark 2.4.0并在pyspark中执行以下代码Python 2.7.16 (default, Mar 25 2019, 15:07:04)>>> from pyspark.sql.functions import pandas_udf>>> from pyspark.

浏览 6提问于2019-03-27得票数 4

回答已采纳

2回答

使用tensorflow.keras模型在pyspark中产生一个泡菜错误

apache-spark、tensorflow、keras、pyspark、user-defined-functions

我想在pysark pandas_udf中使用tensorflow.keras模型。但是，在将模型发送给工作人员之前，当模型被序列化时，我会得到一个泡菜错误。tensorflow.keras.layers import Dense pandas_udf是一个pandas_udf： def compute_output_pandas<

浏览 11提问于2020-04-08得票数 4

回答已采纳

1回答

为什么pandasUDF系列(PandasUDFType.SCALAR_ITER)系列的迭代器(PandasUDFType.SCALAR)是可用的？

apache-spark、pyspark、apache-spark-sql

@pandas_udf('long', PandasUDFType.SCALAR) return v + 1 def pandas_plus_one(iterator): return map(lambdas:

浏览 5提问于2022-01-02得票数 3

回答已采纳

1回答

如何在火花结构化流上使用UDF(用户定义函数)？

pandas、apache-spark、spark-structured-streaming

告诉我，我可以在GroupedData上使用UDF，它可以工作，我可以用自己的函数处理GroupData中的行和列。 window(words.timestamp, "10 minutes", "5 minutes"),).count() 我的问题是是否有一种在words.groupBy

浏览 6提问于2021-03-12得票数 0

回答已采纳

2回答

如何在pyspark pandas_udf中记录/打印消息？

pandas、apache-spark、pyspark、user-defined-functions

我已经测试过logger和print都不能在pandas_udf中打印消息，无论是集群模式还是客户机模式。: ['a'] * 10 + ['b'] *7 + ['q']*3, }) LOGGER = log4jLogge

浏览 0提问于2019-07-24得票数 6

1回答

在阵列上操作的熊猫UDF

pandas、pyspark、apache-spark-sql

我有一个PySpark UDF，它接受一个数组并返回它的后缀：是否有可能将其转化为标量pandas_udf

浏览 0提问于2019-01-30得票数 3

回答已采纳

1回答

Pyspark中的Pandas* Udf在yarn客户端或集群模式下仅在一个执行器中运行*

python、pandas、apache-spark、pyspark

我有一个从Hive Table读取数据并应用pandas udf的代码，当它从表中读取数据时，它在11个执行器中运行，但是当它执行一个pandas udf时，它只使用一个执行器。有没有办法指派10个执行者来执行pandas udf？spark-submit --master yarn --deploy-mode client --conf spark.dynamicAllocation.enabled=false -

浏览 12提问于2020-10-01得票数 0

1回答

将多行结构化流式传输到pandas* udf*

pandas、apache-spark、pyspark、user-defined-functions

我正在写一个结构化的流式作业，它接收来自eventhubs的数据。在做好一些准备之后，我在每一行上应用一个pandas_udf函数来创建一个新列，其中包含来自一个pickle模型的预测。我遇到了一个严重的问题:有时pandas_udf的输入是一组行，而不是单行(正如预期的那样)。这导致了一个错误： RuntimeError: Result vector from pandas_udf was not the requ

浏览 0提问于2019-12-06得票数 1

1回答

有没有办法在Spark中实现矢量化的UDF？

python、pandas、pyspark

在Pandas中，您可以这样做： @pandas_udf(df.schema, PandasUDFType.GROUPED_MAP) returnpdf.assign(v=pdf.v - pdf.v.mean()) df.groupby('id').apply(subtract_mean) 我们有没有办法在Spark中做到这一点？

浏览 26提问于2021-11-09得票数 0

点击加载更多

使用部分函数的pyspark pandas_udf出错

PySpark访问DataFrame列的foreachPartition()自定义函数

pandas_udf给出了与pyarrow相关的错误