Daten im Batch aufnehmen

Die Hauptmethode zum Aufnehmen von Daten in die Manufacturing Data Engine (MDE) ist das Streaming über Manufacturing Connect edge (MCe) oder Senden von Nachrichten an Pub/Sub. Es ist aber auch möglich, Dateien im Batch aufzunehmen. Das kann nützlich sein, um Daten noch einmal zu verarbeiten oder Daten aus externen Systemen zu importieren.

Beim Batch-Laden wird eine IngestionSpecification über die Web oberfläche oder die API erstellt und dann werden eine oder mehrere Dateien in einen Ordner in einem Cloud Storage-Bucket hochgeladen, der von MDE eingerichtet wurde und standardmäßig den Namen <project-id>-batch-ingestion hat. Wenn eine neue Datei in diesen Bucket hochgeladen wird, wird sie vom Dataflow-Job „Cloud Storage Reader“ erkannt. Der Job teilt die Datei in einzelne Nachrichten auf und sendet jede Nachricht an das Pub/Sub-Thema „input-messages“.

Unterstützte Formate

Beim Batch-Laden werden die folgenden Datenformate unterstützt:

  • JSON: Durch Zeilenumbruch getrennt. Jede Zeile wird als neue Nachricht gesendet.
  • CSV: Mit oder ohne Header (kann definiert werden), unterstützt das Überspringen von Zeilen und verschiedene Trennzeichen. Die Nachricht wird in JSON konvertiert, wobei jede Header-Spalte als Schlüssel und die entsprechende Zeilenspalte als Wert verwendet wird.
  • AVRO: Ordnet jede Zeile von AVRO zu JSON zu.
  • AVRO_RAW_WRITER: Liest die unverarbeiteten Nachrichten, die vom Cloud Storage Writer geschrieben wurden, und schreibt jede Nachricht zurück, wobei die Pub/Sub-Nachrichten-ID beibehalten wird.

Konfiguration

Damit der Cloud Storage Reader Dateien verarbeiten kann, benötigt er eine File Ingestion Specification, in der der Dateityp, die Optionen und der Ordner definiert sind, in den die Datei hochgeladen werden soll.

Für alle Formate sind die folgenden Parameter erforderlich:

  • name: Name für die File Ingestion Specification.
  • folderName: Ordner, in den Dateien für diese File Ingestion Specification hochgeladen werden.
  • source: Zu verwendendes Format (JSON/CSV/AVRO/AVRO_RAW_WRITER).

Außerdem werden für die CSV-Quelle die folgenden Parameter unterstützt:

  • separator: Zu verwendendes Trennzeichen (, wird verwendet, wenn keines angegeben ist).
  • skip_rows: Gibt an, ob vor dem Lesen der CSV -Datei Zeilen übersprungen werden sollen.
  • headers:
    • infer_headers: Gibt an, ob die Header aus der ersten gelesenen Zeile abgeleitet werden sollen.
    • headerNames: Geordnete Liste der zu verwendenden Headernamen.
  • insert_metadata: Gibt an, ob Metadaten zur aufgenommenen Datei eingefügt werden sollen (filePath, fileName, ingestionName). Diese werden der gesendeten Nachricht als zusätzliche Schlüssel/Wert-Paare hinzugefügt.

Beispielkonfigurationen

In diesem Abschnitt finden Sie einige Beispielkonfigurationen für das Batch-Laden.

Durch Zeilenumbruch getrennte JSON-Dateien aufnehmen

REST

POST configuration/v1/ingestions

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

Console

  1. Rufen Sie die Konfiguration der File Ingestion Specification auf.

  2. Öffnen Sie im oberen Menü des Bereichs 'Cloud Management' den Abschnitt FILE INGESTION.

    Dateiaufnahme – Dateiaufnahmespezifikationen auflisten

  3. Vorhandene File Ingestion Specifications werden im Abschnitt FILE INGESTION aufgeführt . Sie können über das Symbol Aktionen neben jeder Ingestion Specification geändert, aktiviert, deaktiviert oder gelöscht werden.

    Datei-Aufnahme – Aktion für Aufnahmespezifikationen verfügbar

  4. Klicken Sie auf 'NEUE INGESTION SPECIFICATION HINZUFÜGEN', um eine neue File Ingestion Specification zu erstellen. Rechts auf dem Bildschirm wird ein neues seitliches Menü mit allen erforderlichen Parametern angezeigt, um die neue File Ingestion Specification zu definieren:

    Datei-Ingestion – Neue Spezifikation für die Datei-Ingestion

  5. Wählen Sie im Menü 'Quelltyp' die Option JSON aus, um eine neue Ingestion Specification für durch Zeilenumbruch getrennte JSON-Dateien zu erstellen, und geben Sie die beiden erforderlichen Parameter an. In diesem Beispiel sind das:

    • name: Name für die File Ingestion Specification.
    • folderName: Ordner, in den Dateien für diese File Ingestion Specification hochgeladen werden.

    Datei-Ingestion – Neue Spezifikation für die Ingestion von JSON-Dateien

  6. Klicken Sie nach dem Ausfüllen der Parameter auf Erstellen. Wenn die neue File Ingestion Specification erfolgreich erstellt wurde, wird eine Bestätigungsmeldung angezeigt.

AVRO-Dateien aufnehmen

REST

POST configuration/v1/ingestions

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

Console

  1. Rufen Sie die Konfiguration der File Ingestion Specification auf.

  2. Öffnen Sie im oberen Menü des Bereichs 'Cloud Management' den Abschnitt FILE INGESTION.

    Dateiaufnahme – Dateiaufnahmespezifikationen auflisten

  3. Vorhandene File Ingestion Specifications werden im Abschnitt FILE INGESTION aufgeführt . Sie können über das Symbol Aktionen neben jeder Ingestion Specification geändert, aktiviert, deaktiviert oder gelöscht werden.

    Datei-Aufnahme – Aktion für Aufnahmespezifikationen verfügbar

  4. Klicken Sie auf Neue Ingestion Specification hinzufügen , um eine neue File Ingestion Specification zu erstellen.

  5. Rechts auf dem Bildschirm wird ein neues seitliches Menü mit allen erforderlichen Parametern angezeigt, um die neue File Ingestion Specification zu definieren:

    Datei-Ingestion – Neue Spezifikation für die Datei-Ingestion

  6. Wählen Sie im Menü 'Quelltyp' die Option AVRO aus, um eine neue Ingestion Specification für AVRO-Dateien zu erstellen.

  7. Geben Sie die beiden erforderlichen Parameter an. In diesem Beispiel sind das:

    • name: Name für die File Ingestion Specification.
    • folderName: Ordner, in den Dateien für diese File Ingestion Specification hochgeladen werden.

    Datei-Aufnahme – Neue Spezifikation für die Aufnahme von AVRO-Dateien

  8. Klicken Sie nach dem Ausfüllen der Parameter auf Erstellen. Wenn die neue File Ingestion Specification erfolgreich erstellt wurde, wird eine Bestätigungsmeldung angezeigt.

CSV-Dateien aufnehmen, wobei Spaltennamen aus Headern abgeleitet werden

REST

POST configuration/v1/ingestions

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

Console

  1. Rufen Sie die Weboberfläche von Manufacturing Connect auf.
  2. Rufen Sie die Konfiguration der File Ingestion Specification auf.

  3. Öffnen Sie im oberen Menü des Bereichs 'Cloud Management' den Abschnitt FILE INGESTION.

    Dateiaufnahme – Dateiaufnahmespezifikationen auflisten

  4. Vorhandene File Ingestion Specifications werden im Abschnitt FILE INGESTION aufgeführt . Sie können über das Symbol Aktionen neben jeder Ingestion Specification geändert, aktiviert, deaktiviert oder gelöscht werden.

    Datei-Aufnahme – Aktion für Aufnahmespezifikationen verfügbar

  5. Klicken Sie auf Neue Ingestion Specification hinzufügen , um eine neue File Ingestion Specification zu erstellen.

  6. Rechts auf dem Bildschirm wird ein neues seitliches Menü mit allen erforderlichen Parametern angezeigt, um die neue File Ingestion Specification zu definieren:

    Datei-Ingestion – Neue Spezifikation für die Datei-Ingestion

  7. Klicken Sie auf 'NEUE INGESTION SPECIFICATION HINZUFÜGEN', um eine neue File Ingestion Specification zu erstellen.

  8. Rechts auf dem Bildschirm wird ein neues seitliches Menü mit allen erforderlichen Parametern angezeigt, um die neue **File Ingestion Specification** zu definieren:

    File Ingestion Specification:

    Datei-Ingestion – Neue Spezifikation für die Datei-Ingestion

  9. Wählen Sie im Menü 'Quelltyp&#/1} die Option CSV aus, um eine neue Ingestion Specification für CSV-Dateien zu erstellen, bei der Spaltennamen aus Headern abgeleitet werden, und geben Sie die sieben erforderlichen Parameter an. In diesem Beispiel sind das:

    • name: Name für die File Ingestion Specification.
    • folderName: Ordner, in den Dateien für diese File Ingestion Specification hochgeladen werden.
    • separator: Zu verwendendes Trennzeichen (, wird verwendet, wenn keines angegeben ist).
    • skip_rows: Gibt an, ob vor dem Lesen der CSV -Datei Zeilen übersprungen werden sollen.
    • headers:
    • infer_headers: Gibt an, ob die Header aus der ersten gelesenen Zeile abgeleitet werden sollen. Wählen Sie Ja aus, um Spaltennamen aus Headern abzuleiten.
    • insert_metadata: Gibt an, ob Metadaten zur aufgenommenen Datei eingefügt werden sollen (Dateipfad, Dateiname, IngestionName).

    Dateien aufnehmen – Neue Spezifikation für die Aufnahme von CSV-Dateien

  10. Klicken Sie nach dem Ausfüllen der Parameter auf Erstellen.

  11. Wenn die neue File Ingestion Specification erfolgreich erstellt wurde, wird eine Bestätigungsmeldung angezeigt.

CSV-Dateien aufnehmen, Spaltennamen angeben und Ingestion-Metadaten hinzufügen

REST

POST configuration/v1/ingestions

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

Console

  1. Rufen Sie die Konfiguration der File Ingestion Specification auf.

  2. Öffnen Sie im oberen Menü des Bereichs 'Cloud Management' den Abschnitt FILE INGESTION.

    Dateiaufnahme – Dateiaufnahmespezifikationen auflisten

  3. Vorhandene File Ingestion Specifications werden im Abschnitt FILE INGESTION aufgeführt . Sie können über das Symbol Aktionen neben jeder Ingestion Specification geändert, aktiviert, deaktiviert oder gelöscht werden.

    Datei-Aufnahme – Aktion für Aufnahmespezifikationen verfügbar

  4. Klicken Sie auf Neue Ingestion Specification hinzufügen , um eine neue File Ingestion Specification zu erstellen.

  5. Rechts auf dem Bildschirm wird ein neues seitliches Menü mit allen erforderlichen Parametern angezeigt, um die neue File Ingestion Specification zu definieren:

    Datei-Ingestion – Neue Spezifikation für die Datei-Ingestion

  6. Wählen Sie im Menü 'Quelltyp' die Option CSV aus, um eine neue CSV-Dateien, bei der Spaltennamen angegeben und Ingestion-Metadaten hinzugefügt werden Ingestion Specification zu erstellen.

  7. Geben Sie die acht erforderlichen Parameter an. In diesem Beispiel sind das:

    • name: Name für die File Ingestion Specification.
    • folderName: Ordner, in den Dateien für diese File Ingestion Specification hochgeladen werden.
    • separator: Zu verwendendes Trennzeichen (, wird verwendet, wenn keines angegeben ist).
    • skip_rows: Gibt an, ob vor dem Lesen der CSV -Datei Zeilen übersprungen werden sollen.
    • headers:
      • infer_headers: Gibt an, ob die Header aus der ersten gelesenen Zeile abgeleitet werden sollen. Wählen Sie Nein aus, um Spaltennamen nicht aus Headern abzuleiten.
      • headerNames: Geordnete Liste der zu verwendenden Headernamen.
    • insert_metadata: Gibt an, ob Metadaten zur aufgenommenen Datei eingefügt werden sollen (Dateipfad, Dateiname, IngestionName).

    Dateien aufnehmen – Neue erweiterte Spezifikation für das Aufnehmen von CSV-Dateien

  8. Klicken Sie nach dem Ausfüllen der Parameter auf Erstellen. Wenn die neue File Ingestion Specification erfolgreich erstellt wurde, wird eine Bestätigungsmeldung angezeigt.

CSV-Dateien aufnehmen, Headernamen ableiten und 5 Zeilen überspringen, bevor mit dem Lesen der Datei begonnen wird

REST

POST configuration/v1/ingestions

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

Console

  1. Rufen Sie die Konfiguration der File Ingestion Specification auf.

  2. Öffnen Sie im oberen Menü des Bereichs 'Cloud Management' den Abschnitt FILE INGESTION.

    Dateiaufnahme – Dateiaufnahmespezifikationen auflisten

  3. Vorhandene File Ingestion Specifications werden im Abschnitt FILE INGESTION aufgeführt . Sie können über das Symbol Aktionen neben jeder Ingestion Specification geändert, aktiviert, deaktiviert oder gelöscht werden.

    Datei-Aufnahme – Aktion für Aufnahmespezifikationen verfügbar

  4. Klicken Sie auf Neue Ingestion Specification hinzufügen , um eine neue File Ingestion Specification zu erstellen.

  5. Rechts auf dem Bildschirm wird ein neues seitliches Menü mit allen erforderlichen Parametern angezeigt, um die neue File Ingestion Specification zu definieren:

    Datei-Ingestion – Neue Spezifikation für die Datei-Ingestion

  6. Wählen Sie im Menü 'Quelltyp' die Option CSV aus, um eine neue Ingestion Specification für CSV-Dateien zu erstellen, bei der Headernamen abgeleitet und 5 Zeilen übersprungen werden.

  7. Geben Sie die sieben erforderlichen Parameter an. In diesem Beispiel sind das:

    • name: Name für die File Ingestion Specification.
    • folderName: Ordner, in den Dateien für diese File Ingestion Specification hochgeladen werden.
    • separator: Zu verwendendes Trennzeichen (, wird verwendet, wenn keines angegeben ist).
    • skip_rows: Wählen Sie die Anzahl der zu überspringenden Zeilen aus. In diesem Fall ist das 5.
    • headers:
      • infer_headers: Gibt an, ob die Header aus der ersten gelesenen Zeile abgeleitet werden sollen. Wählen Sie Ja aus, um Spaltennamen aus Headern abzuleiten.
    • insert_metadata: Gibt an, ob Metadaten zur aufgenommenen Datei eingefügt werden sollen (Dateipfad, Dateiname, IngestionName).

    Dateiaufnahme – Neue Spezifikation für die Aufnahme von CSV-Dateien mit übersprungenen Zeilen

  8. Klicken Sie nach dem Ausfüllen der Parameter auf Erstellen.

  9. Wenn die neue File Ingestion Specification erfolgreich erstellt wurde, wird eine Bestätigungsmeldung angezeigt.

AVRO-Dateien aufnehmen, die vom Cloud Storage Writer mit dem Rohpfad generiert wurden

REST

POST configuration/v1/ingestions

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

Das ist besonders nützlich, wenn Sie Dateien noch einmal verarbeiten müssen. Sie können die Datumspartition im Ordner mit den Rohdateien verwenden, um nur die Dateien aus dem benötigten Zeitraum zu kopieren. Beispiel:

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

Console

  1. Rufen Sie die Konfiguration der File Ingestion Specification auf.

  2. Öffnen Sie im oberen Menü des Bereichs 'Cloud Management' den Abschnitt FILE INGESTION.

    Dateiaufnahme – Dateiaufnahmespezifikationen auflisten

  3. Vorhandene File Ingestion Specifications werden im Abschnitt FILE INGESTION aufgeführt . Sie können über das Symbol Aktionen neben jeder Ingestion Specification geändert, aktiviert, deaktiviert oder gelöscht werden.

    Datei-Aufnahme – Aktion für Aufnahmespezifikationen verfügbar

  4. Klicken Sie auf Neue Ingestion Specification hinzufügen , um eine neue File Ingestion Specification zu erstellen.

  5. Rechts auf dem Bildschirm wird ein neues seitliches Menü mit allen erforderlichen Parametern angezeigt, um die neue File Ingestion Specification zu definieren:

    Datei-Ingestion – Neue Spezifikation für die Datei-Ingestion

  6. Wählen Sie im Menü 'Quelltyp' die Option AVRO Raw aus, um eine neue Ingestion Specification für AVRO-Dateien, die vom Cloud Storage Writer mit dem Rohpfad generiert wurden, zu erstellen, und geben Sie die beiden erforderlichen Parameter an. In diesem Beispiel sind das:

    • name: Name für die File Ingestion Specification.
    • folderName: Ordner, in den Dateien für diese File Ingestion Specification hochgeladen werden.

    Datei-Aufnahme – Neue Spezifikation für die Aufnahme von AVRO-Rohdateien

  7. Klicken Sie nach dem Ausfüllen der Parameter auf Erstellen.

  8. Wenn die neue File Ingestion Specification erfolgreich erstellt wurde, wird eine Bestätigungsmeldung angezeigt.

Das ist besonders nützlich, wenn Sie Dateien noch einmal verarbeiten müssen. Sie können die Datumspartition im Ordner mit den Rohdateien verwenden, um nur die Dateien aus dem benötigten Zeitraum zu kopieren. Beispiel:

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