Tengo una canalización de haz de apache simple que lee bigquery >> crea un archivo de parquet en GCS >> Desde la ruta de GCS, leo la información de metadatos y obtengo un objeto JSON con una función ParDo personalizada >> Escribo como texto de regreso a GCS .

No hay ningún error de tiempo de ejecución pero, cuando verifico el archivo creado en la ruta de GCS de destino, tiene 0 bytes . Verifiqué si Pcollection está vacío usando la función beam.ParDo(print) . Cuando hago esto, imprime el objeto JSON. Ya intenté convertir el objeto como str, list y dict. Todo funciona bien con la declaración beam.ParDo(print) pero no puede escribir ningún dato en GCS. Por favor, hágamelo saber, ¿cuál es mi error? Gracias por adelantado :)

 class CreateMetadata(beam.DoFn): def process(self, element, *args, **kwargs): # element here gets a GCS path as gs://..... ### Some internal process ### # This function is called only once at the end of the internal # process have a JSON object as entity data yield entity_data with beam.Pipeline(options=pipeline_options) as pipeline: table = ( pipeline | 'ReadTableFromBigQuery' >> beam.io.Read(beam.io.BigQuerySource(query=bigquery.read_sql(query_string=query), use_standard_sql=True)) | 'writeAsParquetToGCS' >> pq.WriteToParquet(OUTPUT, schema=bigquery.get_parquet_schema_from_bq(), num_shards=1) | 'CreateMetadata' >> beam.ParDo(CreateMetadata()) # | beam.ParDo(print) | 'WriteToGCS' >> beam.io.WriteToText(file_path_prefix=OUTPUT.replace('Some GCS path'), file_name_suffix='.json',num_shards=1) # Now after the pipeline runs when I check the GCS path there is a file created with # 0 bytes??? )```