Передайте словарь в pyspark udf - PullRequest
Новая
       19

Передайте словарь в pyspark udf

0 голосов
/ 29 октября 2018

Я новичок в pyspark и пытаюсь использовать udf для сопоставления некоторых имен строк. Мне нужно сопоставить некоторые значения данных с новыми именами, поэтому я собирался отправить значение столбца из sparkdf и словарь сопоставленных полей в udf вместо того, чтобы писать тонну .when() после .withColumn().

Попытка передать только 2 строки в udf, и это работает, но передача словаря - нет. 

def stringToStr_function(checkCol, dict1) :
  for key, value in dict1.iteritems() :
    if(checkCol != None and checkCol==key): return value

stringToStr_udf = udf(stringToStr_function, StringType())

df = sparkdf.toDF().withColumn(
    "new_col",
     stringToStr_udf(
        lit("REQUEST"),
        {"REQUEST": "Requested", "CONFIRM": "Confirmed", "CANCEL": "Cancelled"}
     )
)

Но получения этой ошибки о методе col () не существует. есть идеи? 

File "<stdin>", line 2, in <module>
  File "/usr/lib/spark/python/pyspark/sql/functions.py", line 1957, in wrapper
    return udf_obj(*args)
  File "/usr/lib/spark/python/pyspark/sql/functions.py", line 1918, in __call__
    return Column(judf.apply(_to_seq(sc, cols, _to_java_column)))
  File "/usr/lib/spark/python/pyspark/sql/column.py", line 60, in _to_seq
    cols = [converter(c) for c in cols]
  File "/usr/lib/spark/python/pyspark/sql/column.py", line 48, in _to_java_column
    jcol = _create_column_from_name(col)
  File "/usr/lib/spark/python/pyspark/sql/column.py", line 41, in _create_column_from_name
    return sc._jvm.functions.col(name)
  File "/usr/lib/spark/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py", line 1133, in __call__
    answer, self.gateway_client, self.target_id, self.name)
  File "/usr/lib/spark/python/pyspark/sql/utils.py", line 63, in deco
    return f(*a, **kw)
  File "/usr/lib/spark/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py", line 323, in get_return_value
    format(target_id, ".", name, value))
Py4JError: An error occurred while calling z:org.apache.spark.sql.functions.col. Trace:

py4j.Py4JException: Method col([class java.util.HashMap]) does not exist
        at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
        at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:339)
        at py4j.Gateway.invoke(Gateway.java:274)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.GatewayConnection.run(GatewayConnection.java:214)
        at java.lang.Thread.run(Thread.java:748)

Спасибо за любую помощь. Я использую клей AWS и Python 2.x, и я тестирую в ноутбуке.

1 Ответ

0 голосов
/ 29 октября 2018

Как показано в связанном дубликате :

Самое чистое решение - передать дополнительные аргументы с помощью замыкания

Однако, вам не нужно udf для этой конкретной проблемы. (См. Функции Spark и производительность UDF? )

Вы можете использовать pyspark.sql.functions.when для реализации IF-THEN-ELSE логики : 

from pyspark.sql.functions import coalesce, col, lit, when

def stringToStr_function(checkCol, dict1):
    return coalesce(
        *[when(col(checkCol) == key, lit(value)) for key, value in dict1.iteritems()]
    )

df = sparkdf.withColumn(
    "new_col",
    stringToStr_function(
        checkCol = lit("REQUEST"),
        dict1 = {"REQUEST": "Requested", "CONFIRM": "Confirmed", "CANCEL": "Cancelled"}
    )
)

Мы перебираем элементы в словаре и используем when, чтобы вернуть value, если значение в checkCol соответствует key. Оберните это в вызове pyspark.sql.functions.coalesce(), который вернет первое ненулевое значение.

...