Блог

Я получаю ошибку при распараллеливании данных с помощью функции parallelize() pyspark. Я использую spark 2.4.3 и python 3.7

 data = [("James","Smith","USA","CA"),
        ("Michael","Rose","USA","NY"),
        ("Robert","Williams","USA","CA"),
        ("Maria","Jones","USA","FL")
      ]
    
columns = ["firstname","lastname","country","state"]
df = spark.createDataFrame(data = data, schema = columns)
# df = spark.createDataFrame(data = data)
    
out = sc.parallelize(df)
        .map(lambda x: (x,1))
        .collect()
  

Вот стек ошибок:

     330                 raise Py4JError(
    331                     "An error occurred while calling {0}{1}{2}. Trace:n{3}n".
--> 332                     format(target_id, ".", name, value))
    333         else:
    334             raise Py4JError(

Py4JError: An error occurred while calling o93.__getstate__. Trace:
py4j.Py4JException: Method __getstate__([]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
    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:238)
    at java.lang.Thread.run(Thread.java:748)
  

Причина, по которой вы получаете эту ошибку, заключается в том, что вы пытаетесь распараллелить фрейм данных pyspark, но вместо этого это должна быть локальная коллекция python (например, список, массив и т. Д.). Вот описание метода parallelize как распространять локальную коллекцию Python для формирования RDD. Рекомендуется использовать xrange, если входные данные представляют диапазон для производительности.

Возможно, вы пытаетесь сделать это вместо этого:

 out = df.rdd
        .map(lambda x: (x,1))
        .collect()