在Pyspark中读取tar.gz压缩文件时使用特定模式过滤文件

1 人关注

我的文件夹 myfolder.tar.gz 里有多个CSV文件。我是这样创建的：首先把我所有的文件放在一个名为 myfolder 的文件夹里，然后准备一个 tar 的文件夹。然后准备好该文件夹的 .gz 。

假设我们有5个文件。

abc_1.csv
abc_2.csv
abc_3.csv
def_1.csv
def_2.csv
我想用Pyspark数据框架过滤读取特定文件名模式的文件。比如我们想把所有abc的文件都读出来。
这不应该给我们def的结果，反之亦然。目前，我可以通过使用spark.read.csv()函数来读取所有的CSV文件。此外，当我使用pathGlobalFilter参数将文件保存在一个简单的文件夹中时，我能够过滤文件，就像这样。
df = spark.read.csv("mypath",pathGlobalFilter="def_[1-9].csv")
但当我能够在tar.gz中做同样的事情时，比如。
df = spark.read.csv("myfolder.tar.gz", pathGlobalFilter="def_[1-9].csv")
我得到了一个错误。
无法推断出CSV的模式。如何从.tar.gz文件中读取。


         
          
           python


         
          
           apache-spark


         
          
           pyspark


         
          
           gzip

tar


          
           
            
             
             
              blackbishop
             
            
            
             发布于
             
             2021-02-05


          
           
            已采纳


          
           
            
             在此基础上
             
              post
             
             你可以把
             
              .tar.gz
             
             文件读成
             
              binaryFile
             
             ，然后用python
             
              tarfile
             
             提取存档成员，并使用regex
             
              def_[1-9]
             
             过滤文件名。其结果是一个rdd，你可以将其转换为一个数据框架。
            
            import re
import tarfile
from io import BytesIO
# extract only the files with which math regex 'def_[1-9].csv'
def extract_files(bytes):
    tar = tarfile.open(fileobj=BytesIO(bytes), mode="r:gz")
    return [tar.extractfile(x).read() for x in tar if re.match(r"def_[1-9].csv", x.name)]
# read binary file and convert to df
rdd = sc.binaryFiles("/path/myfolder.tar.gz") \
        .mapValues(extract_files) \