过滤GCS bucket文件夹中的文件，并删除带有数据流的0字节文件

Question

我目前正在尝试删除谷歌云存储桶中所有0字节的文件。我希望能够用apache beam和一个将在谷歌云项目上运行的数据流运行器来做到这一点。我现在有的是这个(我用< code >隐藏了一些细节

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()

我的一些问题是：

• 这是完成这项任务的正确方法吗
• 如何传递＜code＞apache_beam.io.gcp.gcsfilesystem。GCSFileSystem到DoFn
• 此外，我想删除所有只包含0字节文件的文件夹。我该怎么办

Answer 1

您无需实际读取文件即可检测空文件，只需直接使用FileSystem对象即可检查文件大小并根据需要删除。match（） 函数返回的“文件元数据”对象包括文件的大小。

像这样的东西

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