gcs_transfer.py

import apache_beam as beam
from apache_beam.io import ReadFromText, WriteToText
from apache_beam.io.gcp import gcsio

# Define your GCS source and destination buckets and prefixes
source_bucket = 'your-source-bucket'
destination_bucket = 'your-destination-bucket'
source_prefix = 'source-directory/'
destination_prefix = 'destination-directory/'

# Custom ParDo function to copy objects from source to destination
class CopyObjects(beam.DoFn):
    def process(self, element):
        # element is the path of an object in the source bucket
        source_path = f'gs://{source_bucket}/{element}'
        destination_path = f'gs://{destination_bucket}/{destination_prefix}{element[len(source_prefix):]}'
        
        gcs = gcsio.GcsIO()
        with gcs.open(source_path) as source_file, gcs.open(destination_path, 'w') as destination_file:
            destination_file.write(source_file.read())

# Create a Pipeline
with beam.Pipeline() as p:
    source_objects = (
        p
        | "ListSourceObjects" >> beam.io.gcp.gcsio.ListObjects(source_bucket, prefix=source_prefix)
        | "CopyObjects" >> beam.ParDo(CopyObjects())
    )