Passing a data frame column and external list to udf under withColumn

by

The cleanest solution is to pass additional arguments using closure: 

def make_topic_word(topic_words):
     return udf(lambda c: label_maker_topic(c, topic_words))

df = sc.parallelize([(["union"], )]).toDF(["tokens"])

(df.withColumn("topics", make_topic_word(keyword_list)(col("tokens")))
    .show())

This doesn’t require any changes in keyword_list or the function you wrap with UDF. You can also use this method to pass an arbitrary object. This can be used to pass for example a list of sets for efficient lookups.

If you want to use your current UDF and pass topic_words directly you’ll have to convert it to a column literal first:

from pyspark.sql.functions import array, lit

ks_lit = array(*[array(*[lit(k) for k in ks]) for ks in keyword_list])
df.withColumn("ad", topicWord(col("tokens"), ks_lit)).show()

Depending on your data and requirements there can alternative, more efficient solutions, which don’t require UDFs (explode + aggregate + collapse) or lookups (hashing + vector operations).

More Related Contents:

  1. Applying UDFs on GroupedData in PySpark (with functioning python example)
  2. Using UDF ignores condition in when
  3. How to pass a constant value to Python UDF?
  4. Apache Spark — Assign the result of UDF to multiple dataframe columns
  5. Spark Equivalent of IF Then ELSE
  6. How to add a constant column in a Spark DataFrame?
  7. Load CSV file with Spark
  8. How do I add a new column to a Spark DataFrame (using PySpark)?
  9. How to perform union on two DataFrames with different amounts of columns in spark?
  10. Does spark predicate pushdown work with JDBC?
  11. ‘PipelinedRDD’ object has no attribute ‘toDF’ in PySpark
  12. Count number of non-NaN entries in each column of Spark dataframe with Pyspark
  13. Spark DataFrame: Computing row-wise mean (or any aggregate operation)
  14. Updating a dataframe column in spark
  15. Dividing complex rows of dataframe to simple rows in Pyspark
  16. PySpark converting a column of type ‘map’ to multiple columns in a dataframe
  17. Why is Apache-Spark – Python so slow locally as compared to pandas?
  18. Spark add new column to dataframe with value from previous row
  19. Best way to get the max value in a Spark dataframe column
  20. PySpark create new column with mapping from a dict
  21. Pyspark: Replacing value in a column by searching a dictionary
  22. Apache Spark Python Cosine Similarity over DataFrames
  23. Pivot String column on Pyspark Dataframe
  24. GroupBy column and filter rows with maximum value in Pyspark
  25. Total size of serialized results of 16 tasks (1048.5 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)
  26. Multiple condition filter on dataframe
  27. How to return a “Tuple type” in a UDF in PySpark?
  28. How to connect HBase and Spark using Python?
  29. How to pivot on multiple columns in Spark SQL?
  30. how to add Row id in pySpark dataframes [duplicate]

Leave a Comment