Cómo filtrar según array valor en PySpark?

Te sugerimos que revises esta solución en un ambiente controlado antes de enviarlo a producción, saludos.

Solución:

Para consultas basadas en la igualdad, puede usar array_contains:

df = sc.parallelize([(1, [1, 2, 3]), (2, [4, 5, 6])]).toDF(["k", "v"])
df.createOrReplaceTempView("df")

# With SQL
sqlContext.sql("SELECT * FROM df WHERE array_contains(v, 1)")

# With DSL
from pyspark.sql.functions import array_contains
df.where(array_contains("v", 1))

Si desea utilizar predicados más complejos, tendrá que hacerlo explode o use un UDF, por ejemplo, algo como esto:

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

def exists(f):
    return udf(lambda xs: any(f(x) for x in xs), BooleanType())

df.where(exists(lambda x: x > 3)("v"))

En chispa 2.4. o posterior también es posible utilizar funciones de orden superior

from pyspark.sql.functions import expr

df.where(expr("""aggregate(
    transform(v, x -> x > 3),
    false, 
    (x, y) -> x or y
)"""))

o

df.where(expr("""
    exists(v, x -> x > 3)
"""))

Los contenedores de Python deberían estar disponibles en 3.1 (SPARK-30681).

En Spark 2.4 puedes filtrar array valores usando la función de filtro en la API SQL.

https://spark.apache.org/docs/2.4.0/api/sql/index.html#filter

Aquí hay un ejemplo en pyspark. En el ejemplo, filtramos todos array valores que son cadenas vacías:

df = df.withColumn("ArrayColumn", expr("filter(ArrayColumn, x -> x != '')"))

Reseñas y calificaciones de la guía

---

---

Utiliza Nuestro Buscador