パイプラインを作成するときは、多くの場合、Google Cloud Storage 内のファイルや BigQuery テーブルなど外部ソースからデータを読み取る必要があります。同様に、パイプラインの結果データを、Cloud Storage 内の出力ファイルや BigQuery などの外部データシンクに出力する場合もあります。Dataflow SDK によって提供される変換は、外部ソースのデータを読み取ったり、外部シンクにデータを書き込んだりすることができます。
Dataflow SDK では、一般的な多くのデータ ストレージ タイプに対応する Read 変換と Write 変換が提供されます。また、Read および Write の API は拡張可能です。つまり、組み込みの変換でサポートされないデータ ストレージ形式の読み取りや書き込みをパイプラインで行いたい場合は、独自の Read オペレーションや Write オペレーションを行う機能拡張を構築できます。
入力データを読み取る
Read 変換は、外部ソースからデータを読み取り、パイプラインが使用できるようにそのデータの PCollection 表現を返します。パイプライン作成中のどの時点でも、Read 変換を使用して新しい PCollection を作成できますが、通常はプログラムの最初で行います。
Java
注: Read 変換には入力 PCollection がないため、変換は Pipeline に直接適用されます。通常どおり、apply を呼び出すと、要素がデータを表す、適切な型の PCollection が返されます。詳しくは、パイプラインを作成するをご覧ください。
複数のロケーションから読み取る
Text など多くの Read 変換は、指定する glob 演算子と一致する複数の入力ファイルの読み取りをサポートしています。次の Read 変換の使用方法をご覧ください。glob 演算子（*）を使用して、Google Cloud Storage 内の指定した場所にある、一致するすべての入力ファイルを読み取ります。
Java
p.apply(TextIO.Read.named("ReadFromText")
.from("gs://my_bucket/path/to/input-*.csv");
上記の Read は、Cloud Storage の所定の位置にある、先頭が "input-" で末尾が ".csv" のすべてのファイルを読み取ります。
さまざまなソースのデータを 1 つの PCollection に読み取るには、各ソースを個別に読み取ってから、Flatten 変換を使用して 1 つの PCollection を作成します。
出力データを書き込む
Write 変換は、PCollection のデータを外部データソースに書き込みます。ほとんどの場合、Write 変換をプログラムの最後で使用して、パイプラインの最終結果を出力します。ただし、パイプラインの任意の時点で Write を使用して PCollection のデータを出力することができます。
Write 変換を使用するには、書き込もうとする PCollection に対して apply メソッドを呼び出し、適切な Write 変換を引数として渡します。
Java
PCollection に対して Write 変換の apply を行うと、戻り値は型 PDone のオブジェクトになります。PDone オブジェクトは自明な結果オブジェクトであり、無視してもかまいません。
複数の出力ファイルに書き込む
Text など、ファイルベースの入力データと出力データの場合、Write 変換はデフォルトで複数の出力ファイルに書き込みます。 Cloud Dataflow サービスでは、分割出力ファイルが常に自動的に生成されます。出力ファイル名を Write 変換に渡すと、そのファイル名が、Write 変換によって生成されるすべての出力ファイルの接頭辞として使用されます。
各出力ファイルに接尾辞を付けるには Write 変換に接尾辞を指定します。
Write 変換の次の使用方法をご覧ください。複数の出力ファイルが Cloud Storage 内の場所に書き込まれます。各ファイルの名前は、接頭辞 "numbers"、数値タグ、接尾辞 ".csv" で構成されます。
Java
records.apply(TextIO.Write.named("WriteToText")
.to("gs://my_bucket/path/to/numbers")
.withSuffix(".csv"));
上記の Write は、接頭辞が "numbers" で接尾辞が ".csv" の複数の出力ファイルを Cloud Storage 内の指定のロケーションに書き込みます。
Dataflow SDK に含まれる I/O API
Dataflow SDK にはソースとシンクの API が含まれています。
その他の I/O API
前述した I/O API の他に Dataflow SDK では拡張可能 API が提供され、これを使用すると、独自のカスタム データソースおよびデータシンクを作成することができます。
Java
Dataflow の Source API と Sink API を使用して、独自のカスタム入力ソースとカスタム出力シンクを作成できます。
ご不明な点がありましたら、Google の