是否可以使用 Apache Spark 读取 pdf/音频/视频文件（非结构化数据）？答案

【问题标题】：Is it possible to read pdf/audio/video files(unstructured data) using Apache Spark?是否可以使用 Apache Spark 读取 pdf/音频/视频文件（非结构化数据）？
【发布时间】：2017-12-07 00:09:11
【问题描述】：

是否可以使用 Apache Spark 读取 pdf/音频/视频文件（非结构化数据）？例如，我有数千张 pdf 发票，我想从中读取数据并对其进行一些分析。我必须执行哪些步骤来处理非结构化数据？

【问题讨论】：

搜索reading pdf spark发现blog.cloudera.com/blog/2015/10/…
如果不在 StackOverflow 的主题，则提供站外资源推荐。我已经回答了一些工作流程，您必须自己实施的细节

标签： hadoop apache-spark bigdata

【解决方案1】：

我们有一个场景，我们需要对输入文件使用自定义解密算法。我们不想用 Scala 或 Python 重写该代码。 Python-Spark 代码如下：

from pyspark import SparkContext, SparkConf, HiveContext, AccumulatorParam

def decryptUncompressAndParseFile(filePathAndContents):
    '''each line of the file becomes an RDD record'''
    global acc_errCount, acc_errLog
    proc = subprocess.Popen(['custom_decrypt_program','--decrypt'], 
             stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    (unzippedData, err) = proc.communicate(input=filePathAndContents[1])
    if len(err) > 0:  # problem reading the file
        acc_errCount.add(1)
        acc_errLog.add('Error: '+str(err)+' in file: '+filePathAndContents[0]+
            ', on host: '+ socket.gethostname()+' return code:'+str(returnCode))
        return []  # this is okay with flatMap
    records   = list()
    iterLines = iter(unzippedData.splitlines())
    for line in iterLines:
        #sys.stderr.write('Line: '+str(line)+'\n')
        values = [x.strip() for x in line.split('|')]
        ...
        records.append( (... extract data as appropriate from values into this tuple ...) )
    return records

class StringAccumulator(AccumulatorParam):
    ''' custom accumulator to holds strings '''
    def zero(self,initValue=""):
        return initValue
    def addInPlace(self,str1,str2):
        return str1.strip()+'\n'+str2.strip()

def main():
    ...
    global acc_errCount, acc_errLog
    acc_errCount  = sc.accumulator(0)
    acc_errLog    = sc.accumulator('',StringAccumulator())
    binaryFileTup = sc.binaryFiles(args.inputDir)
    # use flatMap instead of map, to handle corrupt files
    linesRdd = binaryFileTup.flatMap(decryptUncompressAndParseFile, True)
    df = sqlContext.createDataFrame(linesRdd, ourSchema())
    df.registerTempTable("dataTable")
    ...

自定义字符串累加器在识别损坏的输入文件方面非常有用。

【讨论】：

【解决方案2】：

是的，是的。使用sparkContext.binaryFiles 以二进制格式加载文件，然后使用map 将值映射到其他格式 - 例如，使用 Apache Tika 或 Apache POI 解析二进制文件。

伪代码：

val rawFile = sparkContext.binaryFiles(...
val ready = rawFile.map ( here parsing with other framework

重要的是，解析必须使用我之前在回答中提到的其他框架来完成。 Map 会获取 InputStream 作为参数

【讨论】：

我不认为解析必须使用另一个框架来执行。看我的回答。