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 Overflowpython - Filtering pyspark dataframe if text column includes words in specified list - Stack Overflow
Lambda function for filtering RDD in Spark(Python) - check if element not empty string - Stack Overflow
python - Pyspark RDD .filter() with wildcard - Stack Overflow
can I use an if statement with a lambda function?
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)))
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|
+------------------+
writing an if statement with a lambda function:
I'm trying to filter a map in spark using an if statement but get a syntax error. I have not been able to find the error. Can you guys tell me what I am doing wrong here?
full_count_with0val = full_rdd.map(lambda x: json.loads(x[1])).flatMap(lambda x: x['exposures']).map(lambda x: x['pdd_list'] if len(x['pdd_list'])==0).take(5)