df我在 PySpark的資料框中有兩列：
| features | center |
---------- ----------
| [0,1,0] | [1.5,2,1]|
| [5,7,6] | [10,7,7] |
我想創建一個函式來計算和之間的歐幾里得距離，df['features']并將df['center']其映射到 df, 中的新列distance。
假設我們的函式如下所示：
@udf
def dist(feat, cent):
return np.linalg.norm(feat-cent)
我將如何實際應用它來做我想做的事情？我正在嘗試類似的事情
df.withColumn("distance", dist(col("features"), col("center"))).show()
但這給了我以下錯誤：
rg.apache.spark.SparkException：作業因階段故障而中止：階段 869.0 中的任務 0 失敗 4 次，最近一次失敗：階段 869.0 中丟失的任務 0.3（TID 26423）（10.50.91.134 執行程式 35）：net.razorvine。 pickle.PickleException：構造 ClassDict 的預期零引數（用于 numpy.dtype）
我真的很難理解如何在 Spark 背景關系中進行基本的 Python 映射，所以我非常感謝任何幫助。
uj5u.com熱心網友回復：
你真的選擇了一個困難的話題。在 Spark 中，95% 以上的事情可以在沒有 python UDF 的情況下完成。您應該始終嘗試找到不創建 UDF 的方法。
我已經嘗試過您的 UDF，我遇到了同樣的錯誤，但我無法真正說出原因。我認為資料型別有些問題，因為您將 Spark 陣列傳遞給需要 numpy 資料型別的函式。我真的不能說更多...
對于歐幾里得距離，可以在 Spark 中計算。不過，這不是一件容易的事。
from pyspark.sql import functions as F
df = spark.createDataFrame(
[([0, 1, 0], [1.5, 2., 1.]),
([5, 7, 6], [10., 7., 7.])],
['features', 'center'])
distance = F.aggregate(
F.transform(
F.arrays_zip('features', 'center'),
lambda x: (x['features'] - x['center'])**2
),
F.lit(0.0),
lambda acc, x: acc x,
lambda x: x**.5
)
df = df.withColumn('distance', distance)
df.show()
# --------- ---------------- ------------------
# | features| center| distance|
# --------- ---------------- ------------------
# |[0, 1, 0]| [1.5, 2.0, 1.0]|2.0615528128088303|
# |[5, 7, 6]|[10.0, 7.0, 7.0]|5.0990195135927845|
# --------- ---------------- ------------------
uj5u.com熱心網友回復：
from sklearn.metrics.pairwise import paired_distances
更改 dfs 架構以適應 dist 列
sch= df.withColumn('dist', lit(90.087654623)).schema
創建計算距離的 pandas udf
def euclidean_dist(iterator: Iterator[pd.DataFrame]) -> Iterator[pd.DataFrame]:
for pdf in iterator:
yield pdf.assign(dist=paired_distances(pdf['features'].to_list(),pdf['center'].to_list()))
df.mapInPandas(euclidean_dist, schema=sch).show()
解決方案
--------- ---------------- ------------------
| features| center| dist|
--------- ---------------- ------------------
|[0, 1, 0]| [1.5, 2.0, 1.0]|2.0615528128088303|
|[5, 7, 6]|[10.0, 7.0, 7.0]|5.0990195135927845|
--------- ---------------- ------------------
uj5u.com熱心網友回復：
您可以僅使用 PySpark 和 spark sql API 計算距離：
import pyspark.sql.functions as f
df = (
df
.withColumn('distance', f.sqrt(f.expr('aggregate(transform(features, (element, idx) -> pow(element - element_at(center, idx 1), 2)), cast(0 as double), (acc, val) -> acc val)')))
)
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/497517.html
標籤：数据框 阿帕奇火花 pyspark apache-spark-sql 用户定义函数
下一篇：如何鏈接爆炸和結構欄位選擇？