我正在使用Apache光束管道将csv文件从一个容器传输到Azure存储中的另一个容器,并且能够成功传输文件,但目标文件中的记录不是按顺序排列的。下面是我用来传输文件的代码。
String format = LocalDateTime.now().format(DateTimeFormatter.ofPattern("YYYY_MM_DD_HH_MM_SS3")).toString();
String connectionString = "<<AZURE_STORAGE_CONNECTION_STRING>>";
PipelineOptions options = PipelineOptionsFactory.create();
options.as(BlobstoreOptions.class).setAzureConnectionString(connectionString);
Pipeline p = Pipeline.create(options);
p.apply(TextIO.read().from("azfs://storageaccountname/containername/CSVSample.csv"))
.apply("",FileIO.<String>write().to("azfs://storageaccountname/containername/"+format+"/").withNumShards(1).withSuffix(".csv")
.via(TextIO.sink()));
p.run().waitUntilFinish();
Beam PCollection不是有序的。如果您需要顺序,您可以按键分组(在其中强加顺序)并对值集进行排序,将其作为包含换行符的单个字符串元素发出。