Googleのクラウドストレージにアップロードすると出力データがランダムに表示される

Question

google-dataflow-sdkを使用してGoogleクラウドストレージにCSVファイルをアップロードしています。 Googleのクラウドプロジェクトにファイルをアップロードすると、データはクラウド上のランダムな順序でファイルに表示されます。 csvの各行は正しいですが、行はすべての場所にあります。

csv）のヘッダーです。アトリビュート、アトリビュート、アトリビュート）は常に別の行にあります。私はもう一度ストレスを感じ、各列のデータは問題なく、ランダムに配置された行だけです。ここ

は、最初にデータを読み込むコードです：

PCollection<String> csvData = pipeline.apply(TextIO.Read.named("ReadItems") 
              .from(filename)); 

、これはGoogleのクラウドプロジェクトに書き込むコードです：任意の助け

csvData.apply(TextIO.Write.named("WriteToCloud") 
          .to("gs://dbm-poc/"+partnerId+"/"+dateOfReport+modifiedFileName) 
          .withSuffix(".csv")); 

感謝。まず

Answer 1

Graham Polleyが提供する答えが正しいことに同意しますが、データを順序付けられた方法で書くためにはるかに簡単な方法を見つけることができました。私が代わりに私は、クラウド上に必要となるファイルを格納するためにGoogleのクラウド・ストレージ・ライブラリを使用

、そのよう：

public static String writeFile(byte[] content, String filename, String partnerId, String dateOfReport) { 
    Storage storage = StorageOptions.defaultInstance().service(); 
    BlobId blobId = BlobId.of("dbm-poc", partnerId + "/" + dateOfReport + "-" + filename + ".csv"); 
    BlobInfo blobInfo = BlobInfo.builder(blobId).contentType("binary/octet-stream").build(); 
    storage.create(blobInfo, content); 

    return filename; 
} 

public static byte[] readFile(String filename) throws IOException { 
    return Files.readAllBytes(Paths.get(filename)); 
} 

が相互に連携してこれらの2つの方法を使用して、私がいないだけでアップロードすることができましたバケツへのファイルは内容の順序を失うことなく望みましたが、アップロードされたファイルの形式をテキストからバイナリ/オクテットストリームファイルに変更することもできました。つまり、アクセスとダウンロードが可能です。

この方法でも、データをアップロードするためのパイプラインが不要になっているようです。

Answer 2

、あなたのヘッダーの使用を修正する：たとえば

public static TextIO.Write.Bound<String> withHeader(@Nullable String header) 

https://cloud.google.com/dataflow/java-sdk/JavaDoc/com/google/cloud/dataflow/sdk/io/TextIO.Write#withHeader-java.lang.String-

を：

... 
TextIO.Write.withHeader("<header>").apply(..) 
... 

第二に、データフローは、現在Sinksへの書き込みソート/命じサポートしていません。これは主に、分散型/パラレル型アーキテクチャーによるものです。あなたが本当にしたい場合は、独自のカスタムSinkを書くことができます。詳細は、類似の質問hereを参照してください。