תבנית להמרת פורמט קובץ (Avro, ‏ Parquet, ‏ CSV)

תבנית הצינור File Format Conversion (המרת פורמט קובץ) היא צינור אצווה שממיר קבצים שמאוחסנים ב-Cloud Storage מפורמט נתמך אחד לפורמט נתמך אחר.

המערכת תומכת בהמרות הבאות של פורמטים:

• CSV to Avro
• ‫CSV ל-Parquet
• Avro to Parquet
• ‫Parquet ל-Avro

הדרישות לגבי צינורות עיבוד נתונים

• הקטגוריה של Cloud Storage שרוצים לייצא אליה תוכן צריכה להתקיים לפני שמריצים את צינור הנתונים.

פרמטרים של תבניות

פרמטר	תיאור
inputFileFormat	פורמט קובץ הקלט. הערך חייב להיות אחד מהערכים הבאים: [csv, avro, parquet].
outputFileFormat	פורמט קובץ הפלט. הערך חייב להיות אחד מהערכים הבאים: [avro, parquet].
inputFileSpec	תבנית הנתיב של Cloud Storage לקובצי קלט. לדוגמה, gs://bucket-name/path/*.csv
outputBucket	תיקיית Cloud Storage שבה ייכתבו קובצי הפלט. הנתיב הזה חייב להסתיים בלוכסן. לדוגמה, gs://bucket-name/output/
schema	הנתיב ב-Cloud Storage לקובץ סכימת Avro. לדוגמה, gs://bucket-name/schema/my-schema.avsc
containsHeaders	(אופציונלי) קובצי ה-CSV של הקלט מכילים רשומת כותרת (true/false). ערך ברירת המחדל הוא false. נדרש רק כשקוראים קובצי CSV.
csvFormat	(אופציונלי) מפרט פורמט ה-CSV שבו יש להשתמש לניתוח רשומות. ערך ברירת המחדל הוא Default. פרטים נוספים זמינים במאמר בנושא פורמט CSV של Apache Commons.
delimiter	(אופציונלי) התו שמפריד בין השדות בקובצי ה-CSV של הקלט.
outputFilePrefix	(אופציונלי) הקידומת של קובץ הפלט. ערך ברירת המחדל הוא output.
numShards	(אופציונלי) מספר הרסיסים של קובץ הפלט.

הרצת התבנית

המסוף

1. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

   רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר בנושא מיקומי Dataflow.

4. בתפריט הנפתח תבנית של העברת נתונים, בוחרים בתבנית המרת פורמטים של קבצים.
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

gcloud

הערה: כדי להשתמש ב-Google Cloud CLI להפעלת תבניות flex צריך להשתמש בגרסה 284.0.0 או בגרסה מאוחרת יותר של Google Cloud CLI.

במעטפת או בטרמינל, מריצים את התבנית:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/ \
    --parameters \
inputFileFormat=INPUT_FORMAT,\
outputFileFormat=OUTPUT_FORMAT,\
inputFileSpec=INPUT_FILES,\
schema=SCHEMA,\
outputBucket=OUTPUT_FOLDER

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית האב ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫INPUT_FORMAT: פורמט הקובץ של קובץ הקלט. הערך חייב להיות אחד מהערכים הבאים: [csv, avro, parquet]
• ‫OUTPUT_FORMAT: פורמט הקובץ של קובצי הפלט. הערך חייב להיות אחד מהערכים הבאים: [avro, parquet]
• ‫INPUT_FILES: דפוס הנתיב של קובצי הקלט
• ‫OUTPUT_FOLDER: תיקיית Cloud Storage לקובצי פלט
• ‫SCHEMA: הנתיב לקובץ סכימת Avro

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת HTTP POST. מידע נוסף על ה-API ועל היקפי ההרשאות שלו זמין במאמר projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "inputFileFormat": "INPUT_FORMAT",
          "outputFileFormat": "OUTPUT_FORMAT",
          "inputFileSpec": "INPUT_FILES",
          "schema": "SCHEMA",
          "outputBucket": "OUTPUT_FOLDER"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/",
   }
}

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית האב ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫INPUT_FORMAT: פורמט הקובץ של קובץ הקלט. הערך חייב להיות אחד מהערכים הבאים: [csv, avro, parquet]
• ‫OUTPUT_FORMAT: פורמט הקובץ של קובצי הפלט. הערך חייב להיות אחד מהערכים הבאים: [avro, parquet]
• ‫INPUT_FILES: דפוס הנתיב של קובצי הקלט
• ‫OUTPUT_FOLDER: תיקיית Cloud Storage לקובצי פלט
• ‫SCHEMA: הנתיב לקובץ סכימת Avro

קוד המקור של התבנית

Java

/*
 * Copyright (C) 2019 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates;

import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.metadata.TemplateParameter.TemplateEnumOption;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.templates.FileFormatConversion.FileFormatConversionOptions;
import com.google.cloud.teleport.v2.transforms.AvroConverters.AvroOptions;
import com.google.cloud.teleport.v2.transforms.CsvConverters.CsvPipelineOptions;
import com.google.cloud.teleport.v2.transforms.ParquetConverters.ParquetOptions;
import java.util.EnumMap;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link FileFormatConversion} pipeline takes in an input file, converts it to a desired format
 * and saves it to Cloud Storage. Supported file transformations are:
 *
 * <ul>
 *   <li>Csv to Avro
 *   <li>Csv to Parquet
 *   <li>Avro to Parquet
 *   <li>Parquet to Avro
 * </ul>
 *
 * <p><b>Pipeline Requirements</b>
 *
 * <ul>
 *   <li>Input file exists in Google Cloud Storage.
 *   <li>Google Cloud Storage output bucket exists.
 * </ul>
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/file-format-conversion/README_File_Format_Conversion.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "File_Format_Conversion",
    category = TemplateCategory.UTILITIES,
    displayName = "Convert file formats between Avro, Parquet & CSV",
    description = {
      "The File Format Conversion template is a batch pipeline that converts files stored on Cloud Storage from one supported format to another.\n",
      "The following format conversions are supported:\n"
          + "- CSV to Avro\n"
          + "- CSV to Parquet\n"
          + "- Avro to Parquet\n"
          + "- Parquet to Avro"
    },
    optionsClass = FileFormatConversionOptions.class,
    optionalOptions = {"deadletterTable"},
    flexContainerName = "file-format-conversion",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/file-format-conversion",
    contactInformation = "https://cloud.google.com/support",
    requirements = {"The output Cloud Storage bucket must exist before running the pipeline."})
public class FileFormatConversion {

  /** Logger for class. */
  private static final Logger LOG = LoggerFactory.getLogger(FileFormatConversionFactory.class);

  private static EnumMap<ValidFileFormats, String> validFileFormats =
      new EnumMap<ValidFileFormats, String>(ValidFileFormats.class);

  /**
   * The {@link FileFormatConversionOptions} provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface FileFormatConversionOptions
      extends PipelineOptions, CsvPipelineOptions, AvroOptions, ParquetOptions {
    @TemplateParameter.Enum(
        order = 1,
        enumOptions = {
          @TemplateEnumOption("avro"),
          @TemplateEnumOption("csv"),
          @TemplateEnumOption("parquet")
        },
        description = "File format of the input files.",
        helpText = "File format of the input files. Needs to be either avro, parquet or csv.")
    @Required
    String getInputFileFormat();

    void setInputFileFormat(String inputFileFormat);

    @TemplateParameter.Enum(
        order = 2,
        enumOptions = {@TemplateEnumOption("avro"), @TemplateEnumOption("parquet")},
        description = "File format of the output files.",
        helpText = "File format of the output files. Needs to be either avro or parquet.")
    @Required
    String getOutputFileFormat();

    void setOutputFileFormat(String outputFileFormat);
  }

  /** The {@link ValidFileFormats} enum contains all valid file formats. */
  public enum ValidFileFormats {
    CSV,
    AVRO,
    PARQUET
  }

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    FileFormatConversionOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(FileFormatConversionOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline to completion with the specified options.
   *
   * @param options The execution options.
   * @return The pipeline result.
   * @throws RuntimeException thrown if incorrect file formats are passed.
   */
  public static PipelineResult run(FileFormatConversionOptions options) {
    String inputFileFormat = options.getInputFileFormat().toUpperCase();
    String outputFileFormat = options.getOutputFileFormat().toUpperCase();

    validFileFormats.put(ValidFileFormats.CSV, "CSV");
    validFileFormats.put(ValidFileFormats.AVRO, "AVRO");
    validFileFormats.put(ValidFileFormats.PARQUET, "PARQUET");

    if (!validFileFormats.containsValue(inputFileFormat)) {
      throw new IllegalArgumentException("Invalid input file format.");
    }
    if (!validFileFormats.containsValue(outputFileFormat)) {
      throw new IllegalArgumentException("Invalid output file format.");
    }
    if (inputFileFormat.equals(outputFileFormat)) {
      throw new IllegalArgumentException("Input and output file format cannot be the same.");
    }

    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    pipeline.apply(
        inputFileFormat + " to " + outputFileFormat,
        FileFormatConversionFactory.FileFormat.newBuilder()
            .setOptions(options)
            .setInputFileFormat(inputFileFormat)
            .setOutputFileFormat(outputFileFormat)
            .build());

    return pipeline.run();
  }
}

המאמרים הבאים

• מידע נוסף על תבניות Dataflow
• כאן אפשר לעיין ברשימת התבניות ש-Google סיפקה.