Vorlage „Cloud Storage-CSV-Dateien für BigQuery“

Die Pipeline „Cloud Storage-CSV-Dateien für BigQuery“ ist eine Batchpipeline, mit der Sie Daten aus in Cloud Storage gespeicherten CSV-Dateien lesen und das Ergebnis an eine BigQuery-Tabelle anhängen können. Die CSV-Dateien können unkomprimiert oder in einem der Formate komprimiert sein, die auf der Seite CompressionEnum SDK aufgeführt sind.

Pipelineanforderungen

Damit Sie diese Vorlage verwenden können, muss Ihre Pipeline die folgenden Anforderungen erfüllen.

BigQuery-JSON-Schemadatei

Erstellen Sie eine JSON-Datei, die Ihr BigQuery-Schema beschreibt. Das Schema muss ein JSON-Array der obersten Ebene mit dem Namen BigQuery Schema enthalten, dessen Inhalt dem Muster {"name": "COLUMN_NAME", "type": "DATA_TYPE"} folgt.

Die Batchvorlage "Cloud Storage-CSV-Dateien für BigQuery" unterstützt nicht den Import von Daten in Felder des Typs STRUCT (Eintrag) in der BigQuery-Zieltabelle.

Der folgende JSON-Code beschreibt ein BigQuery-Beispielschema:

{
  "BigQuery Schema": [
    {
      "name": "location",
      "type": "STRING"
    },
    {
      "name": "name",
      "type": "STRING"
    },
    {
      "name": "age",
      "type": "STRING"
    },
    {
      "name": "color",
      "type": "STRING"
    },
    {
      "name": "coffee",
      "type": "STRING"
    }
  ]
}

Fehlertabellenschema

Die BigQuery-Tabelle, in der die abgelehnten Datensätze aus CSV-Dateien gespeichert werden, muss mit dem hier definierten Tabellenschema übereinstimmen.

{
  "BigQuery Schema": [
    {
      "name": "RawContent",
      "type": "STRING"
    },
    {
      "name": "ErrorMsg",
      "type": "STRING"
    }
  ]
}

Vorlagenparameter

Erforderliche Parameter

  • inputFilePattern: Der Cloud Storage-Pfad zur CSV-Datei mit dem zu verarbeitenden Text. Beispiel: gs://your-bucket/path/*.csv.
  • schemaJSONPath: Der Cloud Storage-Pfad zur JSON-Datei, die Ihr BigQuery-Schema definiert.
  • outputTable: Der Name der BigQuery-Tabelle, in der die verarbeiteten Daten gespeichert werden. Wenn Sie eine vorhandene BigQuery-Tabelle wiederverwenden, werden die Daten an die Zieltabelle angehängt.
  • bigQueryLoadingTemporaryDirectory: Das temporäre Verzeichnis, das während des BigQuery-Ladevorgangs verwendet werden soll. Beispiel: gs://your-bucket/your-files/temp_dir.
  • badRecordsOutputTable: Der Name der BigQuery-Tabelle, in der die abgelehnten Daten bei der Verarbeitung der CSV-Dateien gespeichert werden sollen. Wenn Sie eine vorhandene BigQuery-Tabelle wiederverwenden, werden die Daten an die Zieltabelle angehängt. Das Schema dieser Tabelle muss mit dem Fehlertabellenschema (https://cloud.google.com/dataflow/docs/guides/templates/provided/cloud-storage-csv-to-bigquery#GcsCSVToBigQueryBadRecordsSchema) übereinstimmen.
  • delimiter: Das Spaltentrennzeichen, das in der CSV-Datei verwendet wird. Beispiel: ,.
  • csvFormat: Das CSV-Format gemäß dem Apache Commons CSV-Format. Die Standardeinstellung ist Default.

Optionale Parameter

  • containsHeaders: Gibt an, ob Header in der CSV-Datei enthalten sind. Die Standardeinstellung ist false.
  • csvFileEncoding: Das Zeichencodierungsformat der CSV-Datei. Zulässige Werte sind US-ASCII, ISO-8859-1, UTF-8 und UTF-16. Standardmäßig ist dies auf UTF8 eingestellt.

Führen Sie die Vorlage aus.

Console

  1. Rufen Sie die Dataflow-Seite Job aus Vorlage erstellen auf.
  2. Zur Seite "Job aus Vorlage erstellen“
  3. Geben Sie im Feld Jobname einen eindeutigen Jobnamen ein.
  4. Optional: Wählen Sie für Regionaler Endpunkt einen Wert aus dem Drop-down-Menü aus. Die Standardregion ist us-central1.

    Eine Liste der Regionen, in denen Sie einen Dataflow-Job ausführen können, finden Sie unter Dataflow-Standorte.

  5. Wählen Sie im Drop-down-Menü Dataflow-Vorlage die Option the CSV files on Cloud Storage to BigQuery (Batch) templateaus.
  6. Geben Sie Ihre Parameterwerte in die Parameterfelder ein.
  7. Klicken Sie auf Job ausführen.

gcloud

Führen Sie die Vorlage in der Shell oder im Terminal aus:

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates-REGION_NAME/VERSION/GCS_CSV_to_BigQuery \
    --region REGION_NAME \
    --parameters \
inputFilePattern=PATH_TO_CSV_DATA,\
schemaJSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
outputTable=BIGQUERY_DESTINATION_TABLE,\
badRecordsOutputTable=BIGQUERY_BAD_RECORDS_TABLE,\
csvFormat=CSV_FORMAT,\
delimiter=DELIMITER,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
containsHeaders=CONTAINS_HEADERS,\
csvFileEncoding=CSV_FILE_ENCODING

Ersetzen Sie dabei Folgendes:

  • PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
  • JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
  • VERSION: Die Version der Vorlage, die Sie verwenden möchten

    Sie können die folgenden Werte verwenden:

    • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
    • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  • REGION_NAME: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
  • PATH_TO_CSV_DATA: der Cloud Storage-Pfad zu Ihren CSV-Dateien
  • PATH_TO_BIGQUERY_SCHEMA_JSON: der Cloud Storage-Pfad zur JSON-Datei, die die Schemadefinition enthält
  • BIGQUERY_DESTINATION_TABLE: Name der BigQuery-Zieltabelle
  • BIGQUERY_BAD_RECORDS_TABLE: der Name der BigQuery-Tabelle mit den fehlerhaften Datensätzen
  • PATH_TO_TEMP_DIR_ON_GCS: der Cloud Storage-Pfad zum temporären Verzeichnis
  • DELIMITER: Trennzeichen der CSV-Datei
  • CSV_FORMAT: CSV-Formatspezifikation zum Parsen von Einträgen
  • CONTAINS_HEADERS: ob die CSV-Dateien Header enthalten
  • CSV_FILE_ENCODING: Codierung in den CSV-Dateien

API

Senden Sie eine HTTP-POST-Anfrage, um die Vorlage mithilfe der REST API auszuführen. Weitere Informationen zur API und ihren Autorisierungsbereichen finden Sie unter projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://dataflow-templates-LOCATION/VERSION/GCS_CSV_to_BigQuery
{
   "jobName": "JOB_NAME",
   "parameters": {
       "inputFilePattern":"PATH_TO_CSV_DATA",
       "schemaJSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "outputTable":"BIGQUERY_DESTINATION_TABLE",
       "badRecordsOutputTable":"BIGQUERY_BAD_RECORDS_TABLE",
       "csvFormat":"CSV_FORMAT",
       "delimiter":"DELIMITER",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
       "containsHeaders": "CONTAINS_HEADERS",
       "csvFileEncoding": "CSV_FILE_ENCODING"
   },
   "environment": { "zone": "us-central1-f" }
}

Ersetzen Sie dabei Folgendes:

  • PROJECT_ID: die ID des Google Cloud-Projekts, in dem Sie den Dataflow-Job ausführen möchten
  • JOB_NAME: ein eindeutiger Jobname Ihrer Wahl
  • VERSION: Die Version der Vorlage, die Sie verwenden möchten

    Sie können die folgenden Werte verwenden:

    • latest zur Verwendung der neuesten Version der Vorlage, die im nicht datierten übergeordneten Ordner im Bucket verfügbar ist: gs://dataflow-templates-REGION_NAME/latest/
    • Den Versionsnamen wie 2023-09-12-00_RC00, um eine bestimmte Version der Vorlage zu verwenden. Diese ist verschachtelt im jeweiligen datierten übergeordneten Ordner im Bucket enthalten: gs://dataflow-templates-REGION_NAME/.
  • LOCATION: die Region, in der Sie Ihren Dataflow-Job bereitstellen möchten, z. B. us-central1
  • PATH_TO_CSV_DATA: der Cloud Storage-Pfad zu Ihren CSV-Dateien
  • PATH_TO_BIGQUERY_SCHEMA_JSON: der Cloud Storage-Pfad zur JSON-Datei, die die Schemadefinition enthält
  • BIGQUERY_DESTINATION_TABLE: Name der BigQuery-Zieltabelle
  • BIGQUERY_BAD_RECORDS_TABLE: der Name der BigQuery-Tabelle mit den fehlerhaften Datensätzen
  • PATH_TO_TEMP_DIR_ON_GCS: der Cloud Storage-Pfad zum temporären Verzeichnis
  • DELIMITER: Trennzeichen der CSV-Datei
  • CSV_FORMAT: CSV-Formatspezifikation zum Parsen von Einträgen
  • CONTAINS_HEADERS: ob die CSV-Dateien Header enthalten
  • CSV_FILE_ENCODING: Codierung in den CSV-Dateien

Nächste Schritte