La tarea de flujo de datos falla porque no puede eliminar el archivo temporal.

DirectRunner hace el trabajo sin problemas, pero Dataflow falla consistentemente porque no puede eliminar y luego no puede renombrar archivos temporales.

Este es el mismo código exacto, mismo bucket, solo cambié los runners:

public static DataflowPipelineOptions setOptions() {
List experiments = new ArrayList(); // experiments debe ser una lista MUTABLE
experiments.add(“use_runner_v2”);

DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);

options.setRunner(DataflowRunner.class);
options.setProject(“{ID DE PROYECTO}”);
options.setJobName(“takt-dataflow-test”);
options.setTempLocation(bucketPath + “temp/”);
options.setStagingLocation(bucketpath + “staging/”);
options.setRegion(region);
options.setServiceAccount(serviceAccountEmail);
options.setExperiments(experiments);
options.setSubnetwork(subNetwork);

return options;
}

public static void main(String[] args) {

String bucketPath = “gs://mypathdelbucket/”;
DataflowPipelineOptions options = setOptions();
Pipeline p = Pipeline.create(options);
String inputPath = bucketPath + “input/test.csv”;
String outputPath = bucketPath + “output/”;
PCollection lines = p.apply(“ReadMyFile”, TextIO.read().from(inputPath));
lines.apply(“WriteMyFile”, TextIO.write().to(outputPath + “test.txt”));

p.run().waitUntilFinish();
}

Las variables no declaradas aquí son para anonimizar cosas que son en realidad strings. Escribo bucketpath directamente en setOptions, por eso no me molesto en pasarlo.

La diferencia entre DirectRunner y DataflowRunner es que comento la línea setRunner. DirectRunner funciona perfecto, Dataflow falla porque no puede eliminar archivos temporales del bucket.

¡Gracias!