Início / coding / Perguntas / 77120311
Accepted
Naveen Balachandran
Asked: 2023-09-17 10:48:24 +0800 CST

Aplicar operação lógica em um dataframe no pyspark

  • 772

Tenho um dataframe com muitas colunas e em uma das colunas tenho a operação lógica que preciso realizar no dataframe. Como exemplo, veja o dataframe abaixo insira a descrição da imagem aqui

Preciso realizar a operação lógica definida na coluna operação lógica nas linhas relevantes

Em um cenário normal, consigo usar expr(). Mas, neste caso, quando quero lê-lo de uma coluna e depois aplicar, ocorre um erro dizendo que a coluna não é iterável.

Alguma sugestão?

pyspark

2 respostas

  1. Best Answer

    Aqui está uma solução usando scala UDF no pyspark, pois eles são mais rápidos que os UDFs python. Você pode encontrar o código para o UDF e o jar de lançamento usado no script pyspark no repositório a seguir.

    Se você quiser fazer modificações na função UDF para atender às suas necessidades futuras, tudo o que você precisa fazer é executar sbt assemblypara compilar o jar.

    Em seguida, chame a com.help.stackoverflow.CheckUDFsclasse do jar para verificar a implementação correta.

    https://github.com/dineshdharme/pyspark-native-udfs

    Código fonte da EvaluateBooleanExpressionclasse:

    package com.help.udf

import org.apache.spark.sql.api.java.UDF3
import org.apache.spark.sql.api.java.UDF4

import scala.reflect.runtime.currentMirror
import scala.tools.reflect.ToolBox


class EvaluateBooleanExpression extends UDF4[Int, Int, Int, String, Boolean] {

  override def call(a_value:Int, b_value:Int, c_value:Int,  given_expression: String): Boolean = {

    var new_expression = given_expression.replaceAll("A", a_value.toString)
    new_expression = new_expression.replaceAll("B", b_value.toString)
    new_expression = new_expression.replaceAll("C", c_value.toString)
    new_expression = new_expression.replaceAll("0", false.toString)
    new_expression = new_expression.replaceAll("1", true.toString)
    //println("Here's the new expression ", new_expression)

    val toolbox = currentMirror.mkToolBox()
    val calc = toolbox.eval(toolbox.parse(new_expression))

    val convertedCalc = calc.toString.toBoolean
    //println("Here's the new expression ", new_expression)

    convertedCalc
  }
}

    Script python Pyspark:

    import sys

from pyspark import SparkContext, SQLContext
import pyspark.sql.functions as F
import pyspark.sql.functions as F
from pyspark import SparkContext, SQLContext
from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("MyApp") \
    .config("spark.jars", "file:/path/to/pyspark-native-udfs/releases/pyspark-native-udfs-assembly-0.1.3.jar") \
    .getOrCreate()

sc = spark.sparkContext
sqlContext = SQLContext(sc)


data1 = [
[0, 1, 1, "(A&B)"],
[1, 1, 1, "(A)"],
[0, 0, 1, "(A|C)"],
      ]

df1Columns = ["A", "B", "C", "exp"]
df1 = sqlContext.createDataFrame(data=data1, schema = df1Columns)
df1 = df1.withColumn("A", F.col("A").cast("int"))
df1 = df1.withColumn("B", F.col("B").cast("int"))
df1 = df1.withColumn("C", F.col("C").cast("int"))

print("Schema of the dataframe")
df1.printSchema()

print("Given dataframe")
df1.show(n=100, truncate=False)



spark.udf.registerJavaFunction("evaluate_boolean_exp_udf", "com.help.udf.EvaluateBooleanExpression", BooleanType())

df1.createOrReplaceTempView("given_table")

df1_array = sqlContext.sql("select *, evaluate_boolean_exp_udf(A, B, C, exp) as bool_exp_evaluated from given_table")
print("Dataframe after applying SCALA NATIVE UDF")
df1_array.show(n=100, truncate=False)

    Saída :

    Schema of the dataframe
root
 |-- A: integer (nullable = true)
 |-- B: integer (nullable = true)
 |-- C: integer (nullable = true)
 |-- exp: string (nullable = true)

Given dataframe
+---+---+---+-----+
|A  |B  |C  |exp  |
+---+---+---+-----+
|0  |1  |1  |(A&B)|
|1  |1  |1  |(A)  |
|0  |0  |1  |(A|C)|
+---+---+---+-----+

Dataframe after applying SCALA NATIVE UDF
+---+---+---+-----+------------------+
|A  |B  |C  |exp  |bool_exp_evaluated|
+---+---+---+-----+------------------+
|0  |1  |1  |(A&B)|false             |
|1  |1  |1  |(A)  |true              |
|0  |0  |1  |(A|C)|true              |
+---+---+---+-----+------------------+
    • 1

  2. Você pode usar a função eval padrão do Python dentro de uma UDF .

    A função espera que os dados estejam em um dict, então primeiro evaltransformamos as colunas de dados em uma estrutura :

    from pyspark.sql import functions as F

eval_udf = F.udf(lambda op, data: eval(op, {}, data.asDict()))

df.withColumn('data', F.struct([df[x] for x in df.columns if x != 'logical_operation'])) \
    .withColumn('result', eval_udf(F.col('logical_operation'), F.col('data'))) \
    .show()

    Saída:

    +---+---+---+-----------------+---------+------+
|  A|  B|  C|logical_operation|     data|result|
+---+---+---+-----------------+---------+------+
|  0|  1|  1|            (A&B)|{0, 1, 1}|     0|
|  1|  1|  1|              (A)|{1, 1, 1}|     1|
|  0|  0|  1|            (A|C)|{0, 0, 1}|     1|
+---+---+---+-----------------+---------+------+

    evalvem com algumas questões de segurança , então verifique se isso pode ser um problema para você!

    • 1

relate perguntas