我有數(shù)據(jù)框，我想使用 pysaprk 中的折線庫(kù)獲取給定地理位置的 lat_long+-----------------+--------------------+----------+ | vid| geolocations| trip_date|+-----------------+--------------------+----------+|58AC21B17LU006754|eurnE||yqU???????...|2020-02-22||2T3EWRFV0LW060632|uocbGfjniOK[Fs@rC...|2020-02-25||JTDP4RCE0LJ014008|w}wtFpdxtM????Q_@...|2020-02-25||4T1BZ1HK8KU029845|}rz_Dp~hhN?@?@???...|2020-03-03|我正在使用 pandas_udf 并且 apache arrow 已啟用from pyspark.sql.functions import col, pandas_udfspark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", "true")lat_long_udf = pandas_udf(lambda geoloc: polyline.decode(geoloc)[0],ArrayType(StringType()))df1=df.withColumn('lat_long',lat_long_udf(df.geolocations))當(dāng)調(diào)用 df.count() 給出結(jié)果時(shí)，但在執(zhí)行 df.show() 時(shí)，我收到如下錯(cuò)誤： 248, in init_stream_yield_batches for series in iterator: File "/Users/prantik.pariksha/opt/anaconda3/lib/python3.8/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 450, in mapper result = tuple(f(*[a[o] for o in arg_offsets]) for (arg_offsets, f) in udfs) File "/Users/prantik.pariksha/opt/anaconda3/lib/python3.8/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 450, in <genexpr> result = tuple(f(*[a[o] for o in arg_offsets]) for (arg_offsets, f) in udfs) File "/Users/prantik.pariksha/opt/anaconda3/lib/python3.8/site-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 110, in <lambda> verify_result_type(f(*a)), len(a[0])), arrow_return_type) File "/Users/prantik.pariksha/opt/anaconda3/lib/python3.8/site-packages/pyspark/python/lib/pyspark.zip/pyspark/util.py", line 107, in wrapper return f(*args, **kwargs)
1 回答
梵蒂岡之花
TA貢獻(xiàn)1900條經(jīng)驗(yàn) 獲得超5個(gè)贊
您很可能會(huì)收到此錯(cuò)誤，因?yàn)?apandas_udf采用 pandas Series 作為輸入，并且您將該decode函數(shù)直接應(yīng)用于該系列，而不是將其應(yīng)用于 pandas Series 中的值。
例如，在下面的示例中，我稍微擴(kuò)展了您的 lambda 函數(shù)，以便您可以看到它。我采用 pandas 系列，將polyline.decode函數(shù)應(yīng)用于該系列，然后再次返回結(jié)果系列。請(qǐng)注意，我還將返回類(lèi)型更改為ArrayType(DoubleType())而不是ArrayType(StringType()).
import pandas as pd
from pyspark.sql.types import ArrayType, DoubleType
....
df = spark.createDataFrame([["~sqU__pR_jpv@_pR"], ["_~t[__pR~qy@_pR"]], ["geolocations"])
@pandas_udf(ArrayType(DoubleType()))
def lat_long_udf(s: pd.Series) -> pd.Series:
return s.apply(lambda x: polyline.decode(x)[0])
df1=df.withColumn('decoded', lat_long_udf(df.geolocations))
df1.collect()
添加回答
舉報(bào)
0/150
提交
取消