过滤 GCS 存储桶文件夹中的文件并使用 Dataflow 删除 0 字节文件

如何解决过滤 GCS 存储桶文件夹中的文件并使用 Dataflow 删除 0 字节文件

我目前正在尝试删除 Google Cloud Storage 存储分区中所有 0 字节的文件。我希望能够使用 apache beam 和一个将运行的数据流运行器来做到这一点一个谷歌云项目。我现在拥有的是这个（我用 <***> 隐藏了一些细节）：

import apache_beam as beam
import apache_beam.io.gcp.gcsfilesystem as gcs
from apache_beam.options.pipeline_options import PipelineOptions

class DetectEmpty(beam.DoFn):
    def process(self,file_path):
        if gfs.size(file_path) == 0:
            yield file_path

def run(argv=None):

    parser = argparse.ArgumentParser()
    parser.add_argument('--input',dest='input',default=<***>,help='<***>')

    options = PipelineOptions()
    google_cloud_options = options.view_as(GoogleCloudOptions)
    google_cloud_options.project = '<***>'
    google_cloud_options.job_name = '<***>'
    options.view_as(StandardOptions).runner = 'DataflowRunner'

    gfs = gcs.GCSFileSystem(pipeline_options)
    p = beam.Pipeline(options=pipeline_options)

    images = p | 'read directory' >> ReadFromText(kNown_args.input)
    empty_images = images | 'discover empty files' >> beam.ParDo(DetectEmpty())

    p.run()

我的一些问题是：

这是完成这项任务的正确方法吗？
如何将 apache_beam.io.gcp.gcsfilesystem.GCSFileSystem 传递给 DoFn？
此外，我想删除所有仅包含 0 字节文件的文件夹。我该怎么做？

解决方法

您不需要实际读取文件来检测空文件，您可以直接使用 FileSystem 对象来检查文件大小并根据需要删除。 match() 函数返回的 FileMetadata 对象包括文件的大小。

类似的东西

class DeleteEmpty(beam.DoFn):
  def __init__(self,gfs):
    self.gfs = gfs

  def process(self,file_metadata):
    if file_metadata.size_in_bytes == 0:
      gfs.delete([file_metadata.path])

files = p | 'Filenames' >> beam.Create(gfs.match([<directory glob pattern>]).metadata_list)
          | 'Reshuffle' >> beam.Reshuffle() # this allows the downstream code to be parallelized after the Create
          | 'Delete empty files' >> beam.ParDo(DeleteEmpty(gfs))

GCS 并没有真正的文件夹；它们只是在使用 UI 或 gsutil 时添加的一种便利。当文件夹中没有对象时，该文件夹就不存在。见https://cloud.google.com/storage/docs/gsutil/addlhelp/HowSubdirectoriesWork