我正在使用 Anaconda、spark 1.3 和 hadoop。我在 hdfs 的特定目录中存储了 xml 文档列表。
我必须使用 python 脚本加载该 xml 文档，以使用 spark 找出重复的文档。
例子:
conf = SparkConf().setAppName("Sample").setMaster("local[*]")
sc = SparkContext(conf=conf)
dir = sc.textFile("hdfs://XXXXXXX")
configfiles = [os.path.join(dirpath, f) for dirpath, dirnames, files in os.walk(dir)for f in files if f.endswith('.xml')]
在这方面我遇到了一些错误:
TypeError: coercing to Unicode: need string or buffer, RDD found
hdfs://xxxxxx MapPartitionsRDD[1] at textFile at NativeMethodAccessorImpl.java:-2
我使用布隆过滤器通过生成哈希值来查找重复项。这在这里不是问题。
通过访问本地存储的文件但不能处理 hdfs 存储的文件。
谁能帮我解决这个问题？
提前致谢
请您参考如下方法:
该错误意味着您正在尝试将非字符串类型(Spark RDD)附加到某处的字符串。如果你阅读 the docs ，您会看到 sc.textFile 正在返回一个 RDD，因此您可能不能将其传递给 os.walk，因为它不是文件路径。您可以尝试调用collect在 RDD 上获取要迭代的列表以传递给 os.walk:
from collections import DefaultDict
#from http://stackoverflow.com/a/3431835/301807
import hashlib
def hashfile(afile, hasher, blocksize=65536):
buf = afile.read(blocksize)
while len(buf) > 0:
hasher.update(buf)
buf = afile.read(blocksize)
return hasher.digest()
dirs = sc.textFile("hdfs://XXXXXXX").collect() #returns a list, not an RDD
configfiles = DefaultDict(list)
for dir in dirs: #for each directory in the list from Spark
for dirpath, dirnames, files in os.walk(dir): #call os.walk with the directory
for f in files if f.endswith('.xml'): #iterate over the files in the directory
path = os.path.join(dirpath, f) #get the file's full path
digest = hashfile(open(path, 'rb'), hashlib.md5()) #hash the contents
configfiles[digest].append(path)
您最终会得到一个将 MD5 总和映射到文件路径的字典。任何具有多个文件路径的 MD5 总和表示这些文件是重复的。这应该只打印重复项:
for md5, paths in configfiles.items() if len(paths) > 1:
print("The following files are duplicates of each other: '" + "', '".join(paths) + "'")
