Your .filter returns an error because it is the sql filter function (expecting a BooleanType() column) on dataframes not the filter function on RDDs. If you want to use the RDD one, just add .rdd:

small_DF.rdd.filter(lambda x: any(word in x.text for word in test_list))

You don't have to use a UDF, you can use regular expressions in pyspark with .rlike on your column "text":

from pyspark.sql import HiveContext
hc = HiveContext(sc)
import pyspark.sql.functions as psf

words = [x.lower() for x in ['starbucks', 'Nvidia', 'IBM', 'Dell']]
data = [['i love Starbucks'],['dell laptops rocks'],['help me I am stuck!']]
df = hc.createDataFrame(data).toDF('text')
df.filter(psf.lower(df.text).rlike('|'.join(words)))
Answer from MaFF on Stack Overflow
🌐
Medium
medium.com › @softwareprocesspains2023 › pyspark-how-to-use-lambda-function-on-spark-dataframe-to-filter-data-37e03fc7d709
Pyspark — How to use lambda function on spark dataframe to filter data | by SoftwareProcessPains2023 | Medium
July 28, 2024 - Pyspark — How to use lambda function on spark dataframe to filter data #import SparkContext from datetime import date from pyspark.sql import SparkSession from pyspark.sql.types import StructField …
🌐
DataCamp
campus.datacamp.com › courses › big-data-fundamentals-with-pyspark › introduction-to-big-data-analysis-with-spark
Use of lambda() with filter() | Spark
In this exercise, you'll be using lambda() function inside the filter() built-in function to find all the numbers divisible by 10 in the list.
Discussions

python - Filtering pyspark dataframe if text column includes words in specified list - Stack Overflow
I've seen questions posted here that are similar to mine, but I'm still getting errors in my code when trying some accepted answers. I have a dataframe with three columns--created _at, text, and w... More on stackoverflow.com
🌐 stackoverflow.com
April 25, 2017
Lambda function for filtering RDD in Spark(Python) - check if element not empty string - Stack Overflow
I have the following RDD 2019-09-24,Debt collection,transworld systems inc. is trying to collect a debt that is not mine not owed and is inaccurate. 2019-09-19,Credit reporting credit repair servic... More on stackoverflow.com
🌐 stackoverflow.com
October 22, 2021
python - Pyspark RDD .filter() with wildcard - Stack Overflow
I have an Pyspark RDD with a text column that I want to use as a a filter, so I have the following code: table2 = table1.filter(lambda x: x[12] == "*TEXT*") To problem is... As you see I'm using ... More on stackoverflow.com
🌐 stackoverflow.com
April 25, 2017
can I use an if statement with a lambda function?
It helps to debug if you split your 1 liner into multiple variable steps at each function. I haven't worked with python but it appears the problem may be in your last map. It appears what you want is to filter for len(x['pdd_list']) == 0 in a filter function. .map(lambda x: x['pdd_list']) . filter(lambda x: len(x['pdd_list'])==0) More on reddit.com
🌐 r/apachespark
4
4
August 12, 2019
🌐
Spark By {Examples}
sparkbyexamples.com › home › python › using filter() with lambda in python
Using filter() with Lambda in Python - Spark By {Examples}
May 31, 2024 - In Python, the filter() function is used to filter elements of an iterable (e.g., a list) based on a certain condition. When combined with the lambda
🌐
Apache
spark.apache.org › docs › latest › api › python › reference › pyspark.sql › api › pyspark.sql.functions.map_filter.html
pyspark.sql.functions.map_filter — PySpark 4.2.0 documentation
>>> from pyspark.sql import functions as sf >>> df = spark.createDataFrame([(1, {"foo": 42.0, "bar": 1.0, "baz": 32.0})], ("id", "data")) >>> row = df.select( ... sf.map_filter("data", lambda k, v: k.startswith("b") & (v > 1.0)).alias("data_filtered") ...
Top answer
1 of 2
5

Your .filter returns an error because it is the sql filter function (expecting a BooleanType() column) on dataframes not the filter function on RDDs. If you want to use the RDD one, just add .rdd:

small_DF.rdd.filter(lambda x: any(word in x.text for word in test_list))

You don't have to use a UDF, you can use regular expressions in pyspark with .rlike on your column "text":

from pyspark.sql import HiveContext
hc = HiveContext(sc)
import pyspark.sql.functions as psf

words = [x.lower() for x in ['starbucks', 'Nvidia', 'IBM', 'Dell']]
data = [['i love Starbucks'],['dell laptops rocks'],['help me I am stuck!']]
df = hc.createDataFrame(data).toDF('text')
df.filter(psf.lower(df.text).rlike('|'.join(words)))
2 of 2
3

I think filter isnt working becuase it expects a boolean output from lambda function and isin just compares with column. You are trying to compare list of words to list of words. Here is something that I tried can give you some direction -

# prepare some test data ==> 

words = [x.lower() for x in ['starbucks', 'Nvidia', 'IBM', 'Dell']]
data = [['i love Starbucks'],['dell laptops rocks'],['help me I am stuck!']]
df = spark.createDataFrame(data).toDF('text')


from pyspark.sql.types import *

def intersect(row):
    # convert each word in lowecase
    row = [x.lower() for x in row.split()]
    return True if set(row).intersection(set(words)) else False


filterUDF = udf(intersect,BooleanType())
df.where(filterUDF(df.text)).show()

output :

+------------------+
|              text|
+------------------+
|  i love Starbucks|
|dell laptops rocks|
+------------------+
🌐
Alpha-epsilon
alpha-epsilon.de › cca175 › 2017 › 09 › 13 › filter-aggregate-join-rank-and-sort-datasets-spark-python
Filter, aggregate, join, rank, and sort datasets (Spark/Python)
from pyspark import SparkContext import re, sys sc = SparkContext("local", "Max Temperature") sc.textFile(sys.argv[1]) \ .map(lambda s: s.split("\t")) \ .filter(lambda rec: (rec[1] != "9999" and re.match("[01459]", rec[2]))) \ .map(lambda rec: (int(rec[0]), int(rec[1]))) \ .reduceByKey(max) \ .saveAsTextFile(sys.argv[2])
🌐
Annefou
annefou.github.io › pyspark › 02-mapreduce
Introduction to big-data using PySpark: Map-filter-Reduce in python
February 26, 2018 - Note that the lambda definition does not include a “return” statement – it always contains a single expression which is returned.
Find elsewhere
🌐
MungingData
mungingdata.com › pyspark › filter-array
Filtering PySpark Arrays and DataFrame Array Columns - MungingData
Use filter to append an arr_evens column that only contains the even numbers from some_arr: from pyspark.sql.functions import * is_even = lambda x: x % 2 == 0 res = df.withColumn("arr_evens", filter(col("some_arr"), is_even)) res.show() +---------------+---------+ | some_arr|arr_evens| +---------------+---------+ |[1, 2, 3, 5, 7]| [2]| | [2, 4, 9]| [2, 4]| +---------------+---------+ The vanilla filter method in Python works similarly: list(filter(is_even, [2, 4, 9])) # [2, 4] The Spark filter function takes is_even as the second argument and the Python filter function takes is_even as the first argument.
🌐
Stack Overflow
stackoverflow.com › questions › 52675628 › filtering-dataframe-in-lambda-function-in-python
apache spark - filtering dataframe in LAMBDA function in python - Stack Overflow
October 6, 2018 - CopyTypeError Traceback (most recent ... washing").first().temp ----> 6 tempx = cloudantdata.filter(lambda x: x[["temperature"]]) 7 ret= tempx.rdd.map(lambda x : pow(x-meanX,2)).sum() 8 print(ret) /usr/local/src/spark21master/spark/python/pyspark/sql/dataframe.py in filter(self, condition) ...
🌐
Ds100
ds100.org › sp18 › assets › lectures › lec26 › Spark.html
Spark
word_counts_by_label = ( records .flatMap(lambda x: ((x['label'], w) for w in x['text'].split())) .filter(lambda x: len(x[1]) > 2) # keep words that have 3 or more letters .map(lambda x: # Count each word (x[1], np.array([1.0, 0.0]) if x[0] == 'spam' else np.array([0.0, 1.0]) )) .reduceByKey(lambda a, b: a + b) # Sum the counts )
🌐
Supergloo Inc
supergloo.com › home › pyspark sql tutorials › mastering pyspark filter: a step-by-step guide through examples
Mastering PySpark Filter: A Step-by-Step Guide through Examples
July 17, 2023 - In PySpark, the DataFrame filter function, filters data together based on specified columns. For example, with a DataFrame containing website click data, we may wish to group together all the platform values contained a certain column.
🌐
Spark Code Hub
sparkcodehub.com › pyspark › rdd › filter
Filter | PYSPARK Tutorial | Spark Code Hub
This works when you’re processing data—like normalizing strings—then filtering based on the transformed values. from pyspark import SparkContext sc = SparkContext("local", "ChainTransform") rdd = sc.parallelize(["cat", "dog", "rat"], 2) mapped_rdd = rdd.map(lambda x: (x, len(x))) filtered_rdd = mapped_rdd.filter(lambda x: x[1] > 3) result = filtered_rdd.collect() print(result) # Output: [] sc.stop() We map ·
🌐
Stack Overflow
stackoverflow.com › questions › 52975338 › filtering-spark-datasets-on-a-single-column-using-a-lambda-function
scala - Filtering Spark Datasets on a single column using a lambda function - Stack Overflow
October 24, 2018 - // The three filteres were broadcasted as a Map // with the key being the name of the filter and value being // the filter itself // dataSet is Dataset[MyEntry] case class MyEntry(col1: Int, col2: Int, col3: Int) val allFilterNames = Array("one","two","three") dataSet.filter(value => { def foo(value: MyEntry): Boolean = { var found = false for (entry <- allFilterNames) { if (broadcastVariable.value.get(entry).get.contains(value.col1)) { return true } } found } foo(value) })
🌐
TutorialsPoint
tutorialspoint.com › pyspark › pyspark_rdd.htm
PySpark - RDD
----------------------------------------filter.py--------------------------------------- from pyspark import SparkContext sc = SparkContext("local", "Filter app") words = sc.parallelize ( ["scala", "java", "hadoop", "spark", "akka", "spark vs hadoop", "pyspark", "pyspark and spark"] ) words_filter = words.filter(lambda x: 'spark' in x) filtered = words_filter.collect() print "Fitered RDD -> %s" % (filtered) ----------------------------------------filter.py---------------------------------------- Command − The command for filter(f) is − ·
🌐
Stack Overflow
stackoverflow.com › questions › 43675600 › cassandra-pyspark-how-to-filter-a-timpstamp-range-using-lambda
apache spark - Cassandra pyspark : how to filter a timpstamp range using lambda - Stack Overflow
currently, I use cassandra pyspark shell to run my python code , I want to filter a time range (e.g. select * from XXXX where time< 2016-08-30 12:00:00+0000 ). from my cassandra database to calculate the max glass_id. I use this line : sc.cassandraTable("poc", "dream").select("glass_id").filter(lambda r:datetime.fromtimestamp( r["glass_start_time"])<datetime(2016, 8, 30, 12, 0, tzinfo=dateutil.tz.tzoffset(None, 0))).map(lambda r: (r["glass_id"], 1)).reduceByKey(lambda a, b: a + b).collect() to filter a time range (e.g.