python - drop - pyspark sql examples

Filtrado de un Pyspark DataFrame con cláusula IN similar a SQL

Question

Quiero filtrar un Pyspark DataFrame con una cláusula IN similar a SQL, como en

sc = SparkContext()
sqlc = SQLContext(sc)
df = sqlc.sql(''SELECT * from my_df WHERE field1 IN a'')

donde a es la tupla (1, 2, 3) . Estoy recibiendo este error:

java.lang.RuntimeException: [1.67] error: `` ('''' esperado pero identificador a encontrado

que básicamente dice que esperaba algo como ''(1, 2, 3)'' en lugar de a. El problema es que no puedo escribir manualmente los valores en un archivo extraído de otro trabajo.

¿Cómo filtraría en este caso?

Answer 1

Cadena que pasa a SQLContext se evaluó en el ámbito del entorno SQL. No captura el cierre. Si desea pasar una variable, tendrá que hacerlo explícitamente utilizando el formato de cadena:

df = sc.parallelize([(1, "foo"), (2, "x"), (3, "bar")]).toDF(("k", "v"))
df.registerTempTable("df")
sqlContext.sql("SELECT * FROM df WHERE v IN {0}".format(("foo", "bar"))).count()
##  2 

Obviamente, esto no es algo que usaría en un entorno SQL "real" debido a consideraciones de seguridad, pero no debería importar aquí.

En la práctica, DataFrame DSL es una gran opción cuando desea crear consultas dinámicas:

from pyspark.sql.functions import col

df.where(col("v").isin({"foo", "bar"})).count()
## 2

Es fácil de construir y componer y maneja todos los detalles de HiveQL / Spark SQL por usted.

Answer 2

Solo una pequeña adición / actualización:

choice_list = ["foo", "bar", "jack", "joan"]

Si desea filtrar el marco de datos "df", de modo que desee mantener las filas basadas en una columna "v" tomando solo los valores de choice_list, entonces

df_filtered = df.where( ( col("v").isin (choice_list) ) )

Answer 3

Un enfoque ligeramente diferente que funcionó para mí es filtrar con una función de filtro personalizada.

def filter_func(a):
"""wrapper function to pass a in udf"""
    def filter_func_(col):
    """filtering function"""
        if col in a.value:
            return True

    return False

return udf(filter_func_, BooleanType())

# Broadcasting allows to pass large variables efficiently
a = sc.broadcast((1, 2, 3))
df = my_df.filter(filter_func(a)(col(''field1''))) /

Answer 4

reiterando lo que @ zero323 ha mencionado anteriormente: podemos hacer lo mismo usando una lista también (no solo la set ) como abajo

from pyspark.sql.functions import col

df.where(col("v").isin(["foo", "bar"])).count()