Spark-缩短名称列表和压缩文件之间的字符串匹配搜索时间

Question

我正在尝试使用Spark在压缩文件列表中搜索字符串列表。以下是我使用的工作代码。使用int键将字符串列表保存到字典中。我正在针对函数fnMatch中的每个文件构建一个以逗号分隔的匹配int键列表。尽管该代码有效，但仍需要几个小时才能完成。哪些优化可以减少运行时间？

#Function to extract zip files 
def zip_extract(x):
    in_memory_data = io.BytesIO(x[1])
    file_obj = zipfile.ZipFile(in_memory_data, "r")
    files = [i for i in file_obj.namelist()]
    return dict(zip(files, [file_obj.open(file).read() for file in files]))


def safeStr(obj):
    try: return str(obj)
    except UnicodeEncodeError:
        return obj.encode('ascii', 'ignore').decode('ascii')
    except: return ""

#Function to match string list contained in dictionary dcust, one by one against file doctext
def fnMatch(doctext,dcust):
  retval=""
  for k in dcust:
     if dcust[k] in doctext:
        retval=retval+","+str(k)

  return retval


schema = StructType([StructField('fpath', StringType(), True),StructField('docText', StringType(), True)])
zips = sc.binaryFiles('hdfs://hp3/user/test/testhdfs/myzipfile.zip')  
files_data = zips.map(zip_extract)
files_data_flat = files_data.flatMap(lambda x: x.items())
files_data_flat_tfm = files_data_flat.map(lambda x: (safeStr(x[0]),safeStr(x[1])))
df = hc.createDataFrame(files_data_flat_tfm,schema)
df2 = df.withColumn("docLength", size_(col("docText"))  ) 


dfcust = hc.sql('select fullname from tbl_custfull').toPandas()
res=len(dfcust)
print "##################################################"+str(res)+"##############################"
dictcust = dfcust.to_dict().values()[0]
strmatches = udf(lambda x: fnMatch(x,dictcust), StringType())

df2 = df2.withColumn("strMatches", strmatches(col("docText"))  ) 

df2.createOrReplaceTempView ("df2")
dfres=hc.sql("SELECT fpath,docLength,strMatches FROM df2 WHERE length(strMatches) >0")
dfres.show(5)

我使用]提交火花作业>

spark-submit \
--conf spark.executor.memory=20g \
--conf spark.executor.cores=5 \
--conf spark.executor.instances=139 \
--conf spark.driver.maxResultSize=8g \
--files /etc/spark2/conf/hive-site.xml \
--master yarn \
--deploy-mode cluster \
myprogram.py
我应该在此处避免某些事情来提高性能吗？尝试更改执行程序的内存和内核，但差异不大。该列表大约有270K字符串，并且有60k文档

我正在尝试使用Spark在压缩文件列表中搜索字符串列表。以下是我使用的工作代码。使用int键将字符串列表保存到字典中。我正在建立一个逗号...

Answer 1

zip扩展名不可拆分，因此无法在记录级别进行并行化。如果可以使用gzip压缩，那么性能将大大提高。另外，将整个文件读入内存哈希图，然后搜索大文件是不必要且不可扩展的。如果使用gzip压缩，则Spark可以自动解压缩并搜索您。这是Scala中的示例。