PySpark: TypeError: l'état devrait être de chaîne ou d'une Colonne

Je suis en train de filtrer un RDD base comme ci-dessous:

spark_df = sc.createDataFrame(pandas_df)
spark_df.filter(lambda r: str(r['target']).startswith('good'))
spark_df.take(5)

Mais les erreurs suivantes:

TypeErrorTraceback (most recent call last)
<ipython-input-8-86cfb363dd8b> in <module>()
      1 spark_df = sc.createDataFrame(pandas_df)
----> 2 spark_df.filter(lambda r: str(r['target']).startswith('good'))
      3 spark_df.take(5)

/usr/local/spark-latest/python/pyspark/sql/dataframe.py in filter(self, condition)
    904             jdf = self._jdf.filter(condition._jc)
    905         else:
--> 906             raise TypeError("condition should be string or Column")
    907         return DataFrame(jdf, self.sql_ctx)
    908 

TypeError: condition should be string or Column

Une idée de ce que j'ai raté? Merci!!!!

Il n'y est une réponse parfaite à droite ci-dessous 😉

OriginalL'auteur Edamame | 2016-10-05

  1. 22

    DataFrame.filter, qui est un alias pour DataFrame.where, s'attend à une expression SQL soit un Column:

    spark_df.filter(col("target").like("good%"))

    ou l'équivalent d'une chaîne SQL:

    spark_df.filter("target LIKE 'good%'")

    Je crois que vous essayez d'utiliser ici RDD.filter qui est complètement différente de la méthode:

    spark_df.rdd.filter(lambda r: r['target'].startswith('good'))

    et ne bénéficie pas de SQL optimisations.

    OriginalL'auteur user6910411

  2. 4

    J'ai été dans cette situation et se sont installés à l'aide d'un UDF:

    from pyspark.sql.functions import udf
from pyspark.sql.types import BooleanType

filtered_df = spark_df.filter(udf(lambda target: target.startswith('good'), 
                                  BooleanType())(spark_df.target))

    Plus lisible serait d'utiliser une définition de fonction normale au lieu de la lambda

    OriginalL'auteur architectonic

  3. -1

    convertir le dataframe en rdd. 

    spark_df = sc.createDataFrame(pandas_df)
spark_df.rdd.filter(lambda r: str(r['target']).startswith('good'))
spark_df.take(5)

    Je pense que ça peut fonctionner!

    OriginalL'auteur Ashok v