pyspark如何遍历broadcast

Posted 2020-08-11 lllkh

因为论文关系要用到pyspark，具体情形如下：

有一个list=[‘aaa‘,‘bbb‘,‘ccc‘,‘ddd‘]，然后有一个rdd内数据类型是str，eg:‘abcdefg‘，正常如果是需要筛选数组包含的rdd数据可以定义一个broadcast，然后写成：

broadcastvalue = sc.broadcast(list)

rdd.filter(lambda x:x in broadcastvalue.value).collect()

我的需求是要筛选str中包含有list中任意一个数据的那些数据,eg:如果str =‘aaaxxxxxx‘，因为list[0]=‘aaa‘ in str，所以这个数据是我需要的，开始时尝试写成：

def choice(data,list):
    for i in list:
        if i in data:
            return True
    return False
broadcastvalue = sc.broadcast(list)
rdd.filter(lambda x:choice(x,broadcastvalue.value)).collect()

但是这样会报错broadcast is not iterable，这是说明broadcast是一个不可迭代的对象，搜索无果后想到了解决方案，竟然不可以迭代那么我就用非迭代的方式遍历就行了：

def choice(data,list):
    for i in range(len(list)):
        if list[i] in data:
            return True
    return False
broadcastvalue = sc.broadcast(list)
rdd.filter(lambda x:choice(x,broadcastvalue.value)).collect()

其实修改很简单，只是不再用它作为一个迭代对象来遍历了。

废话语录：

在做这个的时候python3碰上了许多的UnicodeError问题，解决思路：

1.肯定是编码问题

2.读取数据库的先判断数据库的编码，表编码、字段编码，读取csv的先判断csv的编码

3.再判断python的脚本编码

4.再判断各个字符串的编码情况(str.encode(‘utf-8‘))

最后发现是在存csv的时候字段编码是ascii，然后用.encode(‘utf-8‘)报了UnicodeError错误，最后找到了解决方法.encode(‘utf-8‘,errors=‘ignore‘)