How to pass whole Row to UDF – Spark DataFrame filter

by

You have to use struct() function for constructing the row while making a call to the function, follow these steps.

Import Row,

import org.apache.spark.sql._

Define the UDF 

def myFilterFunction(r:Row) = {r.get(0)==r.get(1)}

Register the UDF 

sqlContext.udf.register("myFilterFunction", myFilterFunction _)

Create the dataFrame

val records = sqlContext.createDataFrame(Seq(("sachin", "sachin"), ("aggarwal", "aggarwal1"))).toDF("text", "text2")

Use the UDF 

records.filter(callUdf("myFilterFunction",struct($"text",$"text2"))).show

When u want all columns to be passed to UDF.

records.filter(callUdf("myFilterFunction",struct(records.columns.map(records(_)) : _*))).show

Result:

+------+------+
|  text| text2|
+------+------+
|sachin|sachin|
+------+------+

More Related Contents:

  1. Spark SQL: apply aggregate functions to a list of columns
  2. Unpivot in spark-sql/pyspark
  3. How to stop INFO messages displaying on spark console?
  4. DataFrame join optimization – Broadcast Hash Join
  5. Overwrite specific partitions in spark dataframe write method
  6. Convert date from String to Date format in Dataframes
  7. What is the meaning of partitionColumn, lowerBound, upperBound, numPartitions parameters?
  8. Spark Transformation – Why is it lazy and what is the advantage?
  9. What is the difference between spark checkpoint and persist to a disk
  10. Xml processing in Spark
  11. What are workers, executors, cores in Spark Standalone cluster?
  12. Spark: Reading files using different delimiter than new line
  13. How to fix ‘TypeError: an integer is required (got type bytes)’ error when trying to run pyspark after installing spark 2.4.4
  14. Spark MLlib LDA, how to infer the topics distribution of a new unseen document?
  15. Why does format(“kafka”) fail with “Failed to find data source: kafka.” (even with uber-jar)?
  16. Multiple Spark applications with HiveContext
  17. How to loop through each row of dataFrame in pyspark
  18. How to group by common element in array?
  19. Understanding Spark serialization
  20. Pyspark: Pass multiple columns in UDF
  21. Spark-Obtaining file name in RDDs
  22. How does Spark aggregate function – aggregateByKey work?
  23. Convert null values to empty array in Spark DataFrame
  24. DataFrame partitionBy to a single Parquet file (per partition)
  25. Temp table caching with spark-sql
  26. PySpark: How to fillna values in dataframe for specific columns?
  27. How to optimize shuffle spill in Apache Spark application
  28. How to calculate Median in spark sqlContext for column of data type double
  29. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’
  30. Understanding Spark’s caching

Leave a Comment