データをバッチで取り込む

Manufacturing Data Engine（MDE）にデータを取り込む主な方法は、Manufacturing Connect エッジ（MCe）を介したストリーミングを使用するか、Pub/Sub にメッセージを送信することですが、ファイルをバッチで取り込むこともできます。これは、データの再処理や外部システムからのデータのインポートに役立ちます。

バッチ読み込みは、ウェブ インターフェースまたは API を使用して IngestionSpecification を作成し、MDE によって設定された Cloud Storage バケット内のフォルダに 1 つ以上のファイルをアップロードすることで機能します。このフォルダのデフォルト名は <project-id>-batch-ingestion です。新しいファイルがこのバケットにアップロードされると、Cloud Storage Reader Dataflow ジョブによって検出されます。このジョブは、ファイルを個々のメッセージに分割し、それぞれを input-messages Pub/Sub トピックに送信します。

サポートされているファイル形式

バッチ読み込みでは、次のデータ形式がサポートされています。

  • JSON: 改行区切り。各行は新しいメッセージとして送信されます。
  • CSV: ヘッダーの有無（定義可能）、行のスキップと異なる区切り文字をサポートします。メッセージは、各ヘッダー列をキー、対応する行の列を値として使用して JSON に変換されます。
  • AVRO: AVRO の各行を JSON にマッピングします。
  • AVRO_RAW_WRITER: Cloud Storage Writer によって書き込まれた未処理のメッセージを読み取り、Pub/Sub メッセージ ID を保持しながら各メッセージを書き戻します。

構成

Cloud Storage Reader がファイルを処理するには、想定されるファイルの種類、オプション、アップロード先のフォルダを定義する File Ingestion Specification が必要です。

すべての形式で次のパラメータが必要です。

  • name: File Ingestion Specification の名前。
  • folderName: この File Ingestion Specification のファイルがアップロードされるフォルダ。
  • source: 使用する形式（JSON/CSV/AVRO/AVRO_RAW_WRITER）。

また、CSV ソースは次の機能をサポートしています。

  • separator: 使用する区切り文字（指定されていない場合は , が使用されます）。
  • skip_rows: CSV ファイルの読み取りを開始する前に、行をスキップするかどうか。
  • headers:
    • infer_headers: 最初の行の読み取りでヘッダーを推測するかどうか。
    • headerNames: 使用するヘッダー名の順序付きリストを指定します。
  • insert_metadata: 取り込まれたファイルに関連するメタデータ（filePath、fileName、ingestionName）を含めるかどうか。これらは、送信されたメッセージに追加の Key-Value ペアとして追加されます。

構成例

このセクションでは、バッチ読み込みの構成例をいくつか示します。

改行区切りの JSON ファイルを取り込む

REST

POST configuration/v1/ingestions

{
  "name": "json-simple",
  "source": "JSON",
  "folderName": "jsonFiles"
}

コンソール

  1. ファイル取り込み仕様の構成にアクセスします。

  2. [Cloud Management] セクションの上部メニューで [FILE INGESTION] セクションを開きます。

    ファイル取り込み - ファイル取り込み仕様を一覧表示する

  3. 既存のファイル取り込み仕様は、[ファイル取り込み] セクションに一覧表示されます。各取り込み仕様の横にある [アクション] アイコンを使用して、変更、有効化、無効化、削除を行うことができます。

    ファイルの取り込み - 取り込み仕様で使用可能なアクション

  4. 新しいファイル取り込み仕様を作成するには、[新しい取り込み仕様を追加] をクリックします。画面の右側に新しいサイドメニューが表示されます。このメニューには、新しいファイル取り込み仕様を定義するために必要なすべてのパラメータが含まれています。

    ファイル取り込み - 新しいファイル取り込み仕様

  5. 新しい改行区切りの JSON ファイルの取り込み仕様を作成するには、[ソースタイプ] メニューから [JSON] を選択し、2 つの必須パラメータを指定します。この例では、次のようになります。

    • name: ファイル取り込み仕様の名前。
    • folderName: このファイル取り込み仕様のファイルがアップロードされるフォルダ。

    ファイルの取り込み - 新しい JSON ファイルの取り込み仕様

  6. パラメータが完了したら、[作成] をクリックします。新しいファイル取り込み仕様が正常に作成されると、確認メッセージが表示されます。

AVRO ファイルを取り込む

REST

POST configuration/v1/ingestions

{
  "name": "avro-simple",
  "source": "AVRO",
  "folderName": "avroFiles"
}

コンソール

  1. ファイル取り込み仕様の構成にアクセスします。

  2. [Cloud Management] セクションの上部メニューで [FILE INGESTION] セクションを開きます。

    ファイル取り込み - ファイル取り込み仕様を一覧表示する

  3. 既存のファイル取り込み仕様は、[ファイル取り込み] セクションに一覧表示されます。各取り込み仕様の横にある [アクション] アイコンを使用して、変更、有効化、無効化、削除を行うことができます。

    ファイルの取り込み - 取り込み仕様で使用可能なアクション

  4. 新しいファイル取り込み仕様を作成するには、[新しい取り込み仕様を追加] をクリックします。

  5. 画面の右側に新しいサイドメニューが表示され、新しいファイル取り込み仕様を定義するために必要なすべてのパラメータが含まれています。

    ファイル取り込み - 新しいファイル取り込み仕様

  6. 新しい AVRO ファイル取り込み仕様を作成するには、[ソースタイプ] メニューから [AVRO] を選択します。

  7. 次の例のように、2 つの必須パラメータを指定します。

    • name: ファイル取り込み仕様の名前。
    • folderName: このファイル取り込み仕様のファイルがアップロードされるフォルダ。

    ファイル取り込み - 新しい AVRO ファイル取り込み仕様

  8. パラメータが完了したら、[作成] をクリックします。新しいファイル取り込み仕様が正常に作成されると、確認メッセージが表示されます。

ヘッダーから列名を推測して CSV ファイルを取り込む

REST

POST configuration/v1/ingestions

{
  "name": "csv-simple",
  "source": "CSV",
  "folderName": "csv-simple",
  "separator": ",",
  "headers": {
    "inferHeaders": true
  }
}

コンソール

  1. Manufacturing Connect ウェブ インターフェースに移動します。
  2. ファイル取り込み仕様の構成にアクセスします。

  3. [Cloud Management] セクションの上部メニューで [FILE INGESTION] セクションを開きます。

    ファイル取り込み - ファイル取り込み仕様を一覧表示する

  4. 既存のファイル取り込み仕様は、[ファイル取り込み] セクションに一覧表示されます。各取り込み仕様の横にある [アクション] アイコンを使用して、変更、有効化、無効化、削除を行うことができます。

    ファイルの取り込み - 取り込み仕様で使用可能なアクション

  5. 新しいファイル取り込み仕様を作成するには、[新しい取り込み仕様を追加] をクリックします。

  6. 画面の右側に新しいサイドメニューが表示され、新しいファイル取り込み仕様を定義するために必要なすべてのパラメータが含まれています。

    ファイル取り込み - 新しいファイル取り込み仕様

  7. 新しいファイル取り込み仕様を作成するには、[新しい取り込み仕様を追加] をクリックします。

  8. 画面の右側に、定義に必要なすべてのパラメータを含む新しいサイドメニューが表示されます。

    ファイル取り込みの仕様:

    ファイル取り込み - 新しいファイル取り込み仕様

  9. ヘッダーから列名を推測する CSV ファイルの新しい取り込み仕様を作成するには、[ソースタイプ] メニューから [CSV] を選択し、7 つの必須パラメータを指定します。この例では、次のようになります。

    • name: ファイル取り込み仕様の名前。
    • folderName: このファイル取り込み仕様のファイルがアップロードされるフォルダ。
    • separator: 使用する区切り文字（指定されていない場合は , が使用されます）。
    • skip_rows: CSV ファイルの読み取りを開始する前に、行をスキップするかどうか。
    • headers:
    • infer_headers: 読み取った最初の行からヘッダーを推測するかどうか。ヘッダーから列名を推測するには、['YES'] を選択します。
    • insert_metadata: 取り込まれたファイルに関連するメタデータ（filePath、fileName、ingestionName）を含めるかどうか。

    ファイル取り込み - 新しい CSV ファイル取り込み仕様

  10. パラメータが完了したら、[作成] をクリックします。

  11. 新しいファイル取り込み仕様が正常に作成されると、確認メッセージが表示されます。

列名を指定して CSV ファイルを取り込み、取り込みメタデータを追加する

REST

POST configuration/v1/ingestions

{
  "name": "csv-headers-metadata",
  "source": "CSV",
  "folderName": "csv-headers",
  "separator": ",",
  "insertMetadata": true,
  "headers": {
    "headerNames": {
      "names": ["one", "two", "three"]
    }
  }
}

コンソール

  1. ファイル取り込み仕様の構成にアクセスします。

  2. [Cloud Management] セクションの上部メニューで [FILE INGESTION] セクションを開きます。

    ファイル取り込み - ファイル取り込み仕様を一覧表示する

  3. 既存のファイル取り込み仕様は、[ファイル取り込み] セクションに一覧表示されます。各取り込み仕様の横にある [アクション] アイコンを使用して、変更、有効化、無効化、削除を行うことができます。

    ファイルの取り込み - 取り込み仕様で使用可能なアクション

  4. 新しいファイル取り込み仕様を作成するには、[新しい取り込み仕様を追加] をクリックします。

  5. 画面の右側に新しいサイドメニューが表示され、新しいファイル取り込み仕様を定義するために必要なすべてのパラメータが含まれています。

    ファイル取り込み - 新しいファイル取り込み仕様

  6. 列名を提供し、取り込みメタデータを追加する CSV ファイルの新しい取り込み仕様を作成するには、[ソースタイプ] から [CSV] を選択します。

  7. 次の 8 つの必須パラメータを指定します。

    • name: ファイル取り込み仕様の名前。
    • folderName: このファイル取り込み仕様のファイルがアップロードされるフォルダ。
    • separator: 使用する区切り文字（指定されていない場合は , が使用されます）。
    • skip_rows: CSV ファイルの読み取りを開始する前に、行をスキップするかどうか。
    • headers:
      • infer_headers: 読み取った最初の行からヘッダーを推測するかどうか。ヘッダーから列名を推測しない場合は、[NO] を選択します。
      • headerNames: 使用するヘッダー名の順序付きリストを指定します。
    • insert_metadata: 取り込まれたファイルに関連するメタデータ（filePath、fileName、ingestionName）を含めるかどうか。

    ファイルの取り込み - 新しい高度な CSV ファイルの取り込み仕様

  8. パラメータが完了したら、[作成] ボタンをクリックします。新しいファイル取り込み仕様が正常に作成されると、確認メッセージが表示されます。

ヘッダー名を推測し、ファイルの読み取りを開始する前に 5 行をスキップして CSV ファイルを取り込む

REST

POST configuration/v1/ingestions

{
  "name": "csv-skip-rows",
  "source": "CSV",
  "folderName": "csv-skip",
  "skipRows": 5,
  "separator": ",",
  "headers": {
    "inferHeaders": true
  }
}

コンソール

  1. ファイル取り込み仕様の構成にアクセスします。

  2. [Cloud Management] セクションの上部メニューで [FILE INGESTION] セクションを開きます。

    ファイル取り込み - ファイル取り込み仕様を一覧表示する

  3. 既存のファイル取り込み仕様は、[ファイル取り込み] セクションに一覧表示されます。各取り込み仕様の横にある [アクション] アイコンを使用して、変更、有効化、無効化、削除を行うことができます。

    ファイルの取り込み - 取り込み仕様で使用可能なアクション

  4. 新しいファイル取り込み仕様を作成するには、[新しい取り込み仕様を追加] をクリックします。

  5. 画面の右側に新しいサイドメニューが表示され、新しいファイル取り込み仕様を定義するために必要なすべてのパラメータが含まれています。

    ファイル取り込み - 新しいファイル取り込み仕様

  6. 新しい CSV ファイルのヘッダー名を推測して 5 行をスキップする取り込み仕様を作成するには、[ソースタイプ] メニューから [CSV] を選択します。

  7. この例では、7 つの必須パラメータを指定します。

    • name: ファイル取り込み仕様の名前。
    • folderName: このファイル取り込み仕様のファイルがアップロードされるフォルダ。
    • separator: 使用する区切り文字（指定されていない場合は , が使用されます）。
    • skip_rows: スキップする行数（この場合は 5）を選択します。
    • headers:
      • infer_headers: 読み取った最初の行からヘッダーを推測するかどうか。ヘッダーから列名を推測するには、['YES'] を選択します。
    • insert_metadata: 取り込まれたファイルに関連するメタデータ（filePath、fileName、ingestionName）を含めるかどうか。

    ファイルの取り込み - 新しいスキップ行 CSV ファイルの取り込み仕様

  8. パラメータが完了したら、[作成] をクリックします。

  9. 新しいファイル取り込み仕様が正常に作成されると、確認メッセージが表示されます。

Cloud Storage Writer の未加工パスで生成された AVRO ファイルを取り込む

REST

POST configuration/v1/ingestions

{
  "name": "avro-reprocess",
  "source": "AVRO_RAW_WRITER",
  "folderName": "avro-raw"
}

これは、ファイルを再処理する必要がある場合に特に便利です。生ファイル フォルダの日付パーティションを使用して、必要な期間のファイルのみをコピーできます。次に例を示します。

gcloud storage cp "gs://<project-id>-raw/v1/dt=2023-06-19/*" \
gs://<project-id>-batch-ingestion/avro-raw/

コンソール

  1. ファイル取り込み仕様の構成にアクセスします。

  2. [Cloud Management] セクションの上部メニューで [FILE INGESTION] セクションを開きます。

    ファイル取り込み - ファイル取り込み仕様を一覧表示する

  3. 既存のファイル取り込み仕様は、[ファイル取り込み] セクションに一覧表示されます。各取り込み仕様の横にある [アクション] アイコンを使用して、変更、有効化、無効化、削除を行うことができます。

    ファイルの取り込み - 取り込み仕様で使用可能なアクション

  4. 新しいファイル取り込み仕様を作成するには、[新しい取り込み仕様を追加] をクリックします。

  5. 画面の右側に新しいサイドメニューが表示され、新しいファイル取り込み仕様を定義するために必要なすべてのパラメータが含まれています。

    ファイル取り込み - 新しいファイル取り込み仕様

  6. Cloud Storage Writer の未加工パスによって生成された新しい AVRO ファイルの取り込み仕様を作成するには、[ソースタイプ] メニューから [AVRO Raw] を選択し、2 つの必須パラメータを指定します。この例では、次のようになります。

    • name: ファイル取り込み仕様の名前。
    • folderName: このファイル取り込み仕様のファイルがアップロードされるフォルダ。

    ファイル取り込み - 新しい AVRO 生ファイル取り込み仕様

  7. パラメータが完了したら、[作成] をクリックします。

  8. 新しいファイル取り込み仕様が正常に作成されると、確認メッセージが表示されます。

これは、ファイルを再処理する必要がある場合に特に便利です。たとえば、生ファイル フォルダの日付パーティションを使用して、必要な期間のファイルのみをコピーできます。

  gcloud storage cp "gs://<project-id>-raw/v1/dt=2023-06-19/*" \
  gs://<project-id>-batch-ingestion/avro-raw/