このページは Cloud Translation API によって翻訳されました。

Dataflow データパイプラインを操作する

概要

Dataflow データパイプラインは、次のタスクに使用できます。

繰り返し実行するジョブのスケジュールを作成する。
複数のジョブ実行でリソースが消費される場所を把握する。
データの更新頻度の目標を定義して管理する。
パイプラインを修正して最適化するために、個々のパイプラインステージまで掘り下げて調査する。

API ドキュメントについては、Data Pipelines リファレンスをご覧ください。

機能

スケジュールに基づいてバッチジョブを実行する繰り返しバッチパイプラインを作成する。
最新バージョンの入力データに対してバッチジョブを繰り返し実行する増分バッチパイプラインを作成する。
パイプラインの概要スコアカードを使用して、パイプラインの総使用量とリソース使用量を確認する。
ストリーミングパイプラインのデータの更新頻度を表示する。この指標は、時間の経過とともに進化し、更新頻度が特定の目標を下回った場合に通知するアラートに関連付けられます。
パイプライン指標グラフを使用して、バッチパイプラインジョブを比較して、異常を見つける。

制限事項

ご利用いただけるリージョン: Cloud Scheduler が利用可能なリージョンでデータパイプラインを作成できます。
割り当て:
- プロジェクトあたりのデフォルトのパイプライン数: 500
- 組織あたりのデフォルトのパイプライン数: 2,500
  
  組織レベルの割り当てはデフォルトで無効になっています。組織レベルの割り当てにオプトインすると、組織ごとにデフォルトで最大 2,500 のパイプラインを作成できます。
注: 割り当ての上限を引き上げる場合は、Google Cloud サポートにご連絡ください。ただし、上限値には Cloud Scheduler と Dataflow の割り当てと上限が適用されます。
ラベル: ユーザー定義のラベルを使用して Dataflow データパイプラインにラベルを付けることはできません。ただし、additionalUserLabels フィールドを使用すると、それらの値が Dataflow ジョブに渡されます。個々の Dataflow ジョブにラベルが適用される仕組みについては、パイプラインオプションをご覧ください。

データパイプラインの種類

Dataflow には、ストリーミングとバッチの 2 種類のデータパイプラインがあります。どちらのパイプラインも、Dataflow テンプレートで定義されているジョブを実行します。

ストリーミングデータパイプライン: ストリーミングデータパイプラインは、Dataflow ストリーミングジョブが作成された直後にそれを実行します。
バッチデータパイプライン: バッチデータパイプラインは、ユーザー定義のスケジュールで Dataflow バッチジョブを実行します。バッチパイプラインの入力ファイル名をパラメータ化することで、増分バッチパイプライン処理が可能になります。

注: バッチデータパイプラインによって作成されたすべての Dataflow バッチジョブ名には、<pipeline_id>-mp--<timestamp>-<random int> という命名規則が使用されます。timestamp の値は秒単位です。文字列 -mp– は、この Dataflow バッチジョブがバッチデータパイプラインによって作成されることを示すために予約されています。pipeline_id は 27 文字を超えると切り捨てられます。

増分バッチパイプライン

日時プレースホルダを使用すると、バッチパイプラインの増分入力ファイルの形式を指定できます。

年、月、日付、時、分、秒のプレースホルダを使用できます。strftime() 形式に従う必要があります。プレースホルダの前には、パーセント記号（%）が付きます。
パイプラインの作成中は、パラメータの形式は検証されません。
- 例: パラメータ化された入力パスとして「gs://bucket/Y」を指定すると、「%」が前に付いていない「Y」は strftime() 形式に対応していないため、「gs://bucket/Y」として評価されます。

スケジュールされたバッチパイプラインの実行時刻ごとに、入力パスのプレースホルダ部分が現在（またはシフト時間）の日時に評価されます。日付の値は、スケジュールされたジョブのタイムゾーンでの現在の日付を使用して評価されます。評価されたパスが入力ファイルのパスと一致する場合、ファイルは、スケジュールされた時間にバッチパイプラインで処理するために取得されます。

例: バッチパイプラインは、PST の各時間 00 分（1 時間毎）に繰り返すようにスケジュールされます。入力パスを gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv としてパラメータ化した場合、2021 年 4 月 15 日午後 6 時（PST）の入力パスは、gs://bucket-name/2021-04-15/prefix-18_00.csv となります。

タイムシフトパラメータを使用する

+/- 分または時間のタイムシフトパラメータを使用できます。パイプラインスケジュールの現在の日時の前後にシフトした評価日時と入力パスの照合をサポートするには、これらのパラメータを中かっこで囲みます。{[+|-][0-9]+[m|h]} という形式を使用します。バッチパイプラインはスケジュールされた時刻に繰り返し実行されますが、入力パスは、指定した時間オフセットで求められます。

例: バッチパイプラインは、PST の各時間 00 分（1 時間毎）に繰り返すようにスケジュールされます。入力パスを gs://bucket-name/%Y-%m-%d/prefix-%H_%M.csv{-2h} としてパラメータ化した場合、2021 年 4 月 15 日午後 6 時（PST）の入力パスは、gs://bucket-name/2021-04-15/prefix-16_00.csv となります。

データパイプラインのロール

Dataflow データパイプラインのオペレーションを正常に実行するには、次に挙げる必須の IAM ロールが必要です。

オペレーションを実行するには適切なロールが必要です。
- Datapipelines.admin: すべてのデータパイプライン操作を実行できます。
- Datapipelines.viewer: データパイプラインとジョブを表示できます。
- Datapipelines.invoker: データパイプラインジョブを実行できます（このロールは API で有効にできます）。
サービスアカウントがユーザー指定か、デフォルトの Compute Engine サービスアカウントかに関係なく、Cloud Scheduler で使用するサービスアカウントには roles/iam.serviceAccountUser ロールが必要です。詳細については、データパイプラインのロールをご覧ください。
そのアカウントに roles/iam.serviceAccountUser ロールが付与されることで Cloud Scheduler と Dataflow が使用するサービスアカウントとして機能できる必要があります。Cloud Scheduler と Dataflow のサービスアカウントを選択しない場合は、デフォルトの Compute Engine サービスアカウントが使用されます。

データパイプラインを作成する

Dataflow データパイプラインは、次の 2 つの方法で作成できます。

ジョブのインポート
データパイプラインの作成

Data Pipelines の設定ページ: Google Cloud コンソールで、Dataflow パイプライン機能に初めてアクセスすると設定ページが開きます。リストにある API を有効にして、データパイプラインを作成します。

ジョブをインポートする

クラシックまたは Flex テンプレートに基づく Dataflow のバッチジョブやストリーミングジョブをインポートして、データパイプラインにできます。

Google Cloud コンソールで、Dataflow の [ジョブ] ページに移動します。

[ジョブ] に移動
完了したジョブを選択し、[ジョブの詳細] ページで [+パイプラインとしてインポート] を選択します。
[テンプレートからのパイプラインの作成] ページのパラメータに、インポートされたジョブのオプションが入力されます。
バッチジョブの場合は、[パイプラインのスケジュール設定] セクションで、繰り返しのスケジュールを指定します。バッチ実行のスケジュール設定で使用される Cloud Scheduler のメールアカウントアドレスは指定しなくても構いません。指定しない場合は、デフォルトの Compute Engine サービスアカウントが使用されます。

データパイプラインを作成する

Google Cloud コンソールで、Dataflow の [データパイプライン] ページに移動します。

[データパイプライン] に移動
[+ データパイプラインを作成] を選択します。
[テンプレートからのパイプラインの作成] ページでパイプライン名を入力し、他のテンプレートの選択フィールドとパラメータフィールドに入力します。
バッチジョブの場合は、[パイプラインのスケジュール設定] セクションで、繰り返しのスケジュールを指定します。バッチ実行のスケジュール設定で使用される Cloud Scheduler のメールアカウントアドレスは指定しなくても構いません。値が指定されていない場合は、デフォルトの Compute Engine サービスアカウントが使用されます。

バッチデータパイプラインを作成する

このサンプルバッチデータパイプラインを作成するには、プロジェクト内の次のリソースにアクセスできる必要があります。

入出力ファイルを保存する Cloud Storage バケット
テーブルを作成する BigQuery データセット。

このパイプラインの例では、Cloud Storage Text to BigQuery バッチパイプラインテンプレートを使用しています。このテンプレートは、Cloud Storage から CSV 形式のファイルを読み取り、変換を実行して、値を 3 つの列を持つ BigQuery テーブルに挿入します。

ローカルドライブに次のファイルを作成します。

宛先 BigQuery テーブルの次のスキーマを含む bq_three_column_table.json ファイル。

{
  "BigQuery Schema": [
    {
      "name": "col1",
      "type": "STRING"
    },
    {
      "name": "col2",
      "type": "STRING"
    },
    {
      "name": "col3",
      "type": "INT64"
    }
  ]
}

BigQuery に挿入する前に入力データの単純な変換を行う split_csv_3cols.js JavaScript ファイル。

function transform(line) {
    var values = line.split(',');
    var obj = new Object();
    obj.col1 = values[0];
    obj.col2 = values[1];
    obj.col3 = values[2];
    var jsonString = JSON.stringify(obj);
    return jsonString;
}

BigQuery テーブルに挿入される複数のレコードを含む file01.csv CSV ファイル。

b8e5087a,74,27531
7a52c051,4a,25846
672de80f,cd,76981
111b92bf,2e,104653
ff658424,f0,149364
e6c17c75,84,38840
833f5a69,8f,76892
d8c833ff,7d,201386
7d3da7fb,d5,81919
3836d29b,70,181524
ca66e6e5,d7,172076
c8475eb6,03,247282
558294df,f3,155392
737b82a8,c7,235523
82c8f5dc,35,468039
57ab17f9,5e,480350
cbcdaf84,bd,354127
52b55391,eb,423078
825b8863,62,88160
26f16d4f,fd,397783

次のように、gcloud storage cp コマンドを使用して、プロジェクトの Cloud Storage バケット内のフォルダにファイルをコピーします。
1. bq_three_column_table.json と split_csv_3cols.js を gs://BUCKET_ID/text_to_bigquery/ にコピーします。
```
gcloud storage cp bq_three_column_table.json gs://BUCKET_ID/text_to_bigquery/
gcloud storage cp split_csv_3cols.js gs://BUCKET_ID/text_to_bigquery/
```
2. file01.csv を gs://BUCKET_ID/inputs/ にコピーします。
```
gcloud storage cp file01.csv gs://BUCKET_ID/inputs/
```
Google Cloud コンソールで、Cloud Storage の [バケット] ページに移動します。

[バケット] に移動
Cloud Storage バケットに tmp フォルダを作成するには、フォルダ名を選択してバケットの詳細ページを開き、[フォルダを作成] をクリックします。
Google Cloud コンソールで、Dataflow の [データパイプライン] ページに移動します。

[データパイプライン] に移動
[データパイプラインを作成] を選択します。[テンプレートからのパイプラインの作成] ページで、次の項目を入力または選択します。
1. [パイプライン名] に「text_to_bq_batch_data_pipeline」と入力します。
2. リージョンエンドポイント: Compute Engine リージョンを選択します。移行元と移行先のリージョンは一致している必要があります。このため、Cloud Storage バケットと BigQuery テーブルは同じリージョン内に存在している必要があります。
3. [Dataflow テンプレート] で、[Process Data in Bulk (batch)] の [Text Files on Cloud Storage to BigQuery] を選択します。
  
  注: [Process Data Continuously (stream)] で同じ名前のストリーミングパイプラインを選択しないでください。
4. パイプラインをスケジュールする: 使用しているタイムゾーンで [Hourly at Minute 25] などのスケジュールを選択します。スケジュールはパイプラインを送信した後に編集できます。バッチ実行のスケジュール設定で使用される Cloud Scheduler のメールアカウントアドレスは指定しなくても構いません。指定しない場合は、デフォルトの Compute Engine サービスアカウントが使用されます。
5. [必須パラメータ] に次のように入力します。
  1. Cloud Storage の JavaScript UDF パス:
```
gs://BUCKET_ID/text_to_bigquery/split_csv_3cols.js
```
  2. JSON パス:
```
BUCKET_ID/text_to_bigquery/bq_three_column_table.json
```
  3. JavaScript UDF 名: transform
  4. BigQuery 出力テーブル:
```
PROJECT_ID:DATASET_ID.three_column_table
```
  5. Cloud Storage 入力パス:
```
BUCKET_ID/inputs/file01.csv
```
  6. 一時的な BigQuery ディレクトリ:
```
BUCKET_ID/tmp
```
  7. 一時的な場所:
```
BUCKET_ID/tmp
```
6. [パイプラインを作成] をクリックします。
パイプラインとテンプレートの情報を確認し、[パイプラインの詳細] ページで現在の値と履歴を確認します。

データパイプラインスケジュールは、[パイプラインの詳細] ページの [パイプラインの情報] パネルで編集できます。

パイプラインスケジュールの横にある編集ボタン。

Dataflow Pipelines コンソールの [実行] ボタンを使用して、バッチパイプラインをオンデマンドで実行することもできます。

サンプルストリーミングデータパイプラインを作成する

サンプルストリーミングデータパイプラインは、サンプルのバッチパイプラインの手順に沿って作成できます。ただし、次の違いがあります。

パイプラインスケジュールでは、ストリーミングデータパイプラインのスケジュールは指定しません。Dataflow ストリーミングジョブはすぐに開始されます。
[Dataflow テンプレート] で、[Process Data Continuously (stream)] の [Text Files on Cloud Storage to BigQuery] を選択します。
ワーカーマシンタイプの場合、パイプラインは gs://BUCKET_ID/inputs/file01.csv パターンに一致する最初のファイルセットを処理します。また、inputs/ フォルダにアップロードしたファイルで、このパターンに一致する追加ファイルも処理します。CSV ファイルのサイズが数 GB を超える場合は、メモリ不足エラーが発生しないように、デフォルトの n1-standard-4 マシンタイプよりも大容量のメモリを備えたマシンタイプ（n1-highmem-8 など）を選択します。

トラブルシューティング

このセクションでは、Dataflow データパイプラインの問題を解決する方法について説明します。

データパイプラインジョブを起動できない

データパイプラインを使用して定期的なジョブスケジュールを作成すると、Dataflow ジョブが起動せず、Cloud Scheduler のログファイルに 503 ステータスエラーが示されることがあります。

この問題は、Dataflow が一時的にジョブを実行できない場合に発生します。

この問題を回避するには、ジョブを再試行するように Cloud Scheduler を構成します。この問題は一時的なものであるため、ジョブを再試行すると成功する可能性があります。Cloud Scheduler での再試行値の設定の詳細については、ジョブを作成するをご覧ください。

パイプラインの目標に対する違反を調査する

以降のセクションでは、パフォーマンス目標を達成していないパイプラインを調査する方法について説明します。

繰り返し実行するバッチパイプライン

パイプラインの健全性の初期分析には、Google Cloud コンソールの [パイプラインの情報] ページで、[個々のジョブのステータス] と [ステップあたりのスレッド時間] グラフを使用します。これらのグラフはパイプラインステータスパネルにあります。

調査例:

1 時間に 1 回（毎 x 時 3 分）に実行される繰り返しバッチパイプラインがあり、通常、各ジョブは約 9 分実行され、すべてのジョブは 10 分未満で完了するという目標が設定しているとします。
ジョブステータスグラフは、ジョブが 10 分以上実行されていたことを表しています。
更新 / 実行履歴のテーブルで、対象の時間に実行されていたジョブを見つけます。クリックして Dataflow ジョブの詳細ページに移動します。そのページで実行ステージが長いジョブを探し、ログでエラーを確認して遅延の原因を特定します。

ストリーミングパイプライン

パイプラインの状態の初期分析には、[パイプラインの詳細] ページの [パイプラインの情報] タブで、データの更新頻度グラフを使用します。このグラフはパイプラインステータスパネルにあります。

調査例:

通常、ストリーミングパイプラインは、20 秒のデータの更新頻度で 1 回出力を生成します。
データ更新頻度の保証目標を 30 秒に設定します。データの更新頻度グラフを確認すると、午前 9 時から 10 時の間のデータの更新頻度が、ほぼ 40 秒に上昇していることがわかります。
[パイプラインの指標] タブに切り替え、さらに分析するためにスループット、CPU 使用率、メモリ使用率のグラフを表示します。

エラー: プロジェクト内にパイプライン ID がすでに存在する

プロジェクト内にすでに存在する名前で新しいパイプラインを作成しようとすると、Pipeline Id already exist within the project というエラーメッセージが表示されます。この問題を回避するために、パイプラインには常に一意の名前を選択してください。

Dataflow データ パイプラインを操作する

概要

機能

制限事項

データ パイプラインの種類

増分バッチ パイプライン

タイムシフト パラメータを使用する

データ パイプラインのロール

データ パイプラインを作成する

ジョブをインポートする

データ パイプラインを作成する

バッチデータ パイプラインを作成する

サンプル ストリーミング データ パイプラインを作成する

トラブルシューティング

データ パイプライン ジョブを起動できない

パイプラインの目標に対する違反を調査する

繰り返し実行するバッチ パイプライン

ストリーミング パイプライン

エラー: プロジェクト内にパイプライン ID がすでに存在する

Dataflow データパイプラインを操作する

データパイプラインの種類

増分バッチパイプライン

タイムシフトパラメータを使用する

データパイプラインのロール

データパイプラインを作成する

データパイプラインを作成する

バッチデータパイプラインを作成する

サンプルストリーミングデータパイプラインを作成する

データパイプラインジョブを起動できない

繰り返し実行するバッチパイプライン

ストリーミングパイプライン