亚洲在线久爱草,狠狠天天香蕉网,天天搞日日干久草,伊人亚洲日本欧美

為了賬號安全,請及時綁定郵箱和手機立即綁定
已解決430363個問題,去搜搜看,總會有你想問的

pandas_udf 給出與 pyarrow 相關的錯誤

pandas_udf 給出與 pyarrow 相關的錯誤

瀟湘沐 2023-07-05 16:15:03
我有數據框,我想使用 pysaprk 中的折線庫獲取給定地理位置的 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))當調用 df.count() 給出結果時,但在執行 df.show() 時,我收到如下錯誤: 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貢獻1900條經驗 獲得超5個贊

您很可能會收到此錯誤,因為 apandas_udf采用 pandas Series 作為輸入,并且您將該decode函數直接應用于該系列,而不是將其應用于 pandas Series 中的值。


例如,在下面的示例中,我稍微擴展了您的 lambda 函數,以便您可以看到它。我采用 pandas 系列,將polyline.decode函數應用于該系列,然后再次返回結果系列。請注意,我還將返回類型更改為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()


查看完整回答
反對 回復 2023-07-05
  • 1 回答
  • 0 關注
  • 151 瀏覽
慕課專欄
更多

添加回答

舉報

0/150
提交
取消
微信客服

購課補貼
聯系客服咨詢優惠詳情

幫助反饋 APP下載

慕課網APP
您的移動學習伙伴

公眾號

掃描二維碼
關注慕課網微信公眾號