He visto esta pregunta respondida antes en el desbordamiento de pila https: //stackoverflow.com/questions/29983621/how-to-get-filename-when-using-file-pattern-match-in-google-cloud-dataflo), pero no desde que Apache Beam ha agregado la funcionalidad Dofn divisible para Python. ¿Cómo accedería al nombre de archivo del archivo actual que se está procesando al pasar un patrón de archivo a un depósito de gcs?

Quiero pasar el nombre del archivo a mi función de transformación:

with beam.Pipeline(options=pipeline_options) as p:                              
    lines = p | ReadFromText('gs://url to file')                                        


    data = (                                                                    
        lines                                                                   
        | 'Jsonify' >> beam.Map(jsonify)                                        
        | 'Unnest' >> beam.FlatMap(unnest)                                      
        | 'Write to BQ' >> beam.io.Write(beam.io.BigQuerySink(                  
            'project_id:dataset_id.table_name', schema=schema,                     
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,    
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)       
        )                                                   

n última instancia, lo que quiero hacer es pasar el nombre del archivo a mi función de transformación cuando transformo cada fila del json (consulteest y luego use el nombre del archivo para buscar en una tabla BQ diferente para obtener un valor). Creo que una vez que logre saber cómo obtener el nombre de archivo, podré averiguar la parte de entrada lateral para realizar la búsqueda en la tabla bq y obtener el valor único.