将字符串列解析为 Pyspark Row

Posted 2023-03-23

【中文标题】将字符串列解析为 Pyspark Row

【英文标题】：Parse string column into Pyspark Row

【发布时间】：2022-01-23 20:01:23

【问题描述】：

我正在使用其中一列包含以下内容的数据集：

'Row(AcceptsInsurance=None, AgesAllowed=None, Alcohol="\'beer_and_wine\'", Ambience="\'touristy\': False, \'hipster\': False, \'romantic\': False, \'divey\': False, \'intimate\': False, \'trendy\': False, \'upscale\': False, \'classy\': False, \'casual\': True", BYOB=None, BYOBCorkage=None, BestNights=None, BikeParking=\'True\', BusinessAcceptsBitcoin=\'False\', BusinessAcceptsCreditCards=\'True\', BusinessParking="\'garage\': False, \'street\': True, \'validated\': False, \'lot\': False, \'valet\': False")'

我环顾四周，显然这是一个 pyspark 行对象，但我不知道如何将其从字符串转换为 Row。

【问题讨论】：

类似from pyspark.sql import Row 然后Row(x[4:-2])

【参考方案1】：

我认为一个好的解决方案是更正生成此数据集的过程，不应将其保存为 Row 对象。

在 Pyspark 中，您可以使用一些字符串函数（split、regexp_extract...）将其解析为多个列，但这可能非常乏味。尤其是该行包含复杂对象，例如Ambience。

您可能考虑的另一种可能性是尝试将 spark 数据帧转换为 pandas 并使用 python eval（但 not recommended）将该字符串评估为 pyspark Row 对象：

import pandas as pd


sdf = spark.createDataFrame([
    ('Row(AcceptsInsurance=None, AgesAllowed=None, Alcohol="\'beer_and_wine\'", Ambience="\'touristy\': False, \'hipster\': False, \'romantic\': False, \'divey\': False, \'intimate\': False, \'trendy\': False, \'upscale\': False, \'classy\': False, \'casual\': True", BYOB=None, BYOBCorkage=None, BestNights=None, BikeParking=\'True\', BusinessAcceptsBitcoin=\'False\', BusinessAcceptsCreditCards=\'True\', BusinessParking="\'garage\': False, \'street\': True, \'validated\': False, \'lot\': False, \'valet\': False")',)
], ["row"])

df = sdf.toPandas()["row"].apply(lambda x: eval(x).asDict()).apply(pd.Series).astype(str)

sdf = spark.createDataFrame(df)

sdf.show()

#+----------------+-----------+---------------+--------------------+----+-----------+----------+-----------+----------------------+--------------------------+--------------------+
#|AcceptsInsurance|AgesAllowed|        Alcohol|            Ambience|BYOB|BYOBCorkage|BestNights|BikeParking|BusinessAcceptsBitcoin|BusinessAcceptsCreditCards|     BusinessParking|
#+----------------+-----------+---------------+--------------------+----+-----------+----------+-----------+----------------------+--------------------------+--------------------+
#|            None|       None|'beer_and_wine'|'touristy': Fals...|None|       None|      None|       True|                 False|                      True|'garage': False,...|
#+----------------+-----------+---------------+--------------------+----+-----------+----------+-----------+----------------------+--------------------------+--------------------+

【讨论】：

使用 eval 是邪恶的。我们同意。但是如果我们要进入那个兔子洞......为什么我们不在pyspark中应用eval作为lambda？你为什么建议我们使用 python/panda 而不是 pyspark/dataframe？ （你得到了什么我没有看到的附加值？）

@MattAndruff 正如我上面提到的，最好的方法可能是更改生成这些可怕数据的代码。当然，这不是一个好的解决方案，但想指出 OP 可以考虑的可能性。关于 eval 与 pyspark RDD map 或 udf 的使用，NameError: name 'Row' is not defined 会失败，如果你有办法，请告诉我:)

同意，修复源是方法。仅供参考 -> 实际上，如果您 import from pyspark.sql import Row 您可以在没有 Pandas 的情况下使用 udf/map。我希望用 Pandas 包装一些如何使事情更安全的方法，但它似乎只是另一种评估方式。

@MattAndruff 你测试过吗？因为如果它像导入一个类一样简单，我会使用它而不是 pandas。但也许我错过了一些愚蠢的东西，所以如果你能展示如何做到这一点，那就太好了。

from pyspark.sql import Row