Eu já vi essa pergunta respondida antes no estouro de pilha https: //stackoverflow.com/questions/29983621/how-to-get-filename-when-using-file-pattern-match-in-google-cloud-dataflo), mas não desde que o apache beam tenha adicionado a funcionalidade dofn divisível para python. Como eu acessaria o nome do arquivo do arquivo atual que está sendo processado ao passar um padrão de arquivo para um bucket do gc

Quero passar o nome do arquivo para minha função de transformação:

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

Finalmente, o que eu quero fazer é passar o nome do arquivo para a minha função de transformação quando eu transformar cada linha do json (consulteist e use o nome do arquivo para fazer uma pesquisa em uma tabela BQ diferente para obter um valor). Acho que depois que eu conseguir saber como obter o nome do arquivo, poderei descobrir a parte da entrada lateral para fazer a pesquisa na tabela bq e obter o valor exclusiv