טעינת נתונים מ-Amazon S3 ל-BigQuery

המסוף

1. נכנסים לדף Data transfers במסוף Google Cloud .

   מעבר אל "העברות נתונים"

2. ‫add Create transfer.

3. בדף Create Transfer:

   • בקטע סוג המקור, בוחרים באפשרות Amazon S3 בשדה מקור.

     

   • בקטע Transfer config name, בשדה השם המוצג מזינים שם להעברה, למשל My Transfer. שם ההעברה יכול להיות כל ערך שמאפשר לזהות את ההעברה אם תצטרכו לשנות אותה בהמשך.

     

   • בקטע אפשרויות תזמון:

     • בוחרים תדירות חזרה. אם בוחרים באפשרות שעות, ימים, שבועות או חודשים, צריך לציין גם תדירות. אפשר גם לבחור באפשרות בהתאמה אישית כדי ליצור תדירות חזרה ספציפית יותר. אם בוחרים באפשרות On-demand, העברת הנתונים תתבצע רק כשמפעילים את ההעברה באופן ידני.

     • אם רלוונטי, בוחרים באפשרות Start now או באפשרות Start at set time, ומציינים תאריך התחלה וזמן ריצה.

   • בקטע הגדרות היעד:

     • בקטע Dataset, בוחרים את מערך הנתונים שיצרתם כדי לאחסן את הנתונים.
     • בוחרים באפשרות Native table אם רוצים להעביר לטבלה ב-BigQuery.
     • בוחרים באפשרות Iceberg Managed אם רוצים להעביר לטבלה מנוהלת של Iceberg.

   • בקטע פרטי מקור הנתונים:

     • בשדה Destination table, מזינים את שם הטבלה שיצרתם כדי לאחסן את הנתונים ב-BigQuery. שמות טבלאות היעד תומכים בפרמטרים.
     • בשדה Amazon S3 URI, מזינים את ה-URI בפורמט s3://mybucket/myfolder/.... מזהי URI תומכים גם בפרמטרים.
     • בשדה Access key ID (מזהה מפתח הגישה), מזינים את מזהה מפתח הגישה.
     • בשדה Secret access key, מזינים את מפתח הגישה הסודי.
     • בשדה File format בוחרים את פורמט הנתונים (newline delimited JSON,‏ CSV,‏ Avro,‏ Parquet או ORC).
     • בשדה Write Disposition, בוחרים אחת מהאפשרויות הבאות:
       • ‫WRITE_APPEND כדי להוסיף נתונים חדשים באופן מצטבר לטבלת היעד הקיימת. ‫WRITE_APPEND הוא ערך ברירת המחדל של העדפת הכתיבה.
       • ‫WRITE_TRUNCATE כדי להחליף את הנתונים בטבלת היעד בכל הפעלה של העברת נתונים.

     במאמר הטמעת נתונים להעברות מ-Amazon S3 תוכלו לקרוא מידע נוסף על האופן שבו שירות העברת הנתונים ל-BigQuery מטמיע נתונים באמצעות WRITE_APPEND או WRITE_TRUNCATE. מידע נוסף על השדה writeDisposition זמין במאמר JobConfigurationLoad.

     

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

     • בשדה Number of errors allowed, מזינים מספר שלם שמייצג את המספר המקסימלי של רשומות לא תקינות שאפשר להתעלם מהן.
     • (אופציונלי) בשדה Decimal target types, מזינים רשימה מופרדת בפסיקים של סוגי נתונים אפשריים ב-SQL שאליהם אפשר להמיר את הערכים העשרוניים של המקור. סוג הנתונים ב-SQL שנבחר להמרה תלוי בתנאים הבאים:
       • סוג הנתונים שנבחר להמרה יהיה סוג הנתונים הראשון ברשימה הבאה שתומך בדיוק ובקנה המידה של נתוני המקור, בסדר הזה: NUMERIC,‏ BIGNUMERIC ו-STRING.
       • אם אף אחד מסוגי הנתונים שברשימה לא תומך ברמת הדיוק ובקנה המידה, נבחר סוג הנתונים שתומך בטווח הרחב ביותר ברשימה שצוינה. אם ערך חורג מהטווח הנתמך בזמן קריאת נתוני המקור, תופיע שגיאה.
       • סוג הנתונים STRING תומך בכל ערכי הדיוק והקנה מידה.
       • אם השדה הזה נשאר ריק, סוג הנתונים יהיה כברירת מחדל "NUMERIC,STRING" ל-ORC, ו-"NUMERIC" לפורמטים האחרים של הקבצים.
       • השדה הזה לא יכול להכיל סוגי נתונים כפולים.
       • המערכת מתעלמת מהסדר של סוגי הנתונים שאתם מפרטים בשדה הזה.

     

   • אם בחרתם בפורמט קובץ CSV או JSON, בקטע JSON,CSV מסמנים את האפשרות Ignore unknown values כדי לקבל שורות שמכילות ערכים שלא תואמים לסכימה. המערכת מתעלמת מערכים לא ידועים. בקובצי CSV, האפשרות הזו מתעלמת מערכים נוספים בסוף השורה.

     

   • אם בחרתם בפורמט קובץ CSV, בקטע CSV מזינים את האפשרויות הנוספות של CSV לטעינת נתונים.

     

   • בתפריט Service Account, בוחרים חשבון שירות מבין חשבונות השירות שמשויכים לפרויקט ב-Google Cloud . אתם יכולים לשייך חשבון שירות להעברת הנתונים במקום להשתמש בפרטי הכניסה של המשתמש. למידע נוסף על השימוש בחשבונות שירות בהעברות נתונים, קראו את המאמר שימוש בחשבונות שירות.

     • אם נכנסתם באמצעות זהות מאוחדת, תצטרכו חשבון שירות כדי ליצור העברת נתונים. אם נכנסתם באמצעות חשבון Google, חשבון שירות להעברת הנתונים הוא אופציונלי.
     • לחשבון השירות צריכות להיות ההרשאות הנדרשות.

   • אם אתם משתמשים ב-CMEK, בקטע Advanced options בוחרים באפשרות Customer-managed key. תופיע רשימה של מפתחות ה-CMEK הזמינים לבחירה.

   • (אופציונלי) בקטע אפשרויות להתראות:

     • לוחצים על המתג כדי להפעיל את ההתראות באימייל. כשמפעילים את האפשרות הזו, האדמין של ההעברה מקבל התראה באימייל אם הרצת העברת הנתונים נכשלת.
     • בקטע Select a Pub/Sub topic, בוחרים את שם הנושא או לוחצים על Create a topic כדי ליצור נושא חדש. האפשרות הזו מגדירה התראות על הפעלות של Pub/Sub להעברת הנתונים.

4. לוחצים על Save.

BQ

מזינים את הפקודה bq mk ומציינים את הדגל ליצירת העברה – --transfer_config.

bq mk \
--transfer_config \
--project_id=project_id \
--data_source=data_source \
--display_name=name \
--target_dataset=dataset \
--service_account_name=service_account \
--destination_kms_key=destination_key \
--params='parameters'

כאשר:

• ‫project_id: אופציונלי. מזהה הפרויקט ב- Google Cloud . אם לא מציינים פרויקט מסוים באמצעות --project_id, המערכת משתמשת בפרויקט שמוגדר כברירת מחדל.
• ‫data_source: חובה. מקור הנתונים – amazon_s3.
• ‫display_name: חובה. השם המוצג של הגדרת העברת הנתונים. שם ההעברה יכול להיות כל ערך שמאפשר לכם לזהות את ההעברה אם תצטרכו לשנות אותה בהמשך.
• ‫dataset: חובה. מערך הנתונים שאליו מועברים הנתונים.
• ‫service_account: השם של חשבון השירות שמשמש לאימות של העברת הנתונים. חשבון השירות צריך להיות בבעלות של אותו project_id ששימש ליצירת העברת הנתונים, וצריכות להיות לו כל ההרשאות הנדרשות.
• ‫destination_key: אופציונלי. מזהה המשאב של מפתח Cloud KMS – לדוגמה, projects/project_name/locations/us/keyRings/key_ring_name/cryptoKeys/key_name.

• ‫parameters: חובה. הפרמטרים של הגדרת ההעברה שנוצרה בפורמט JSON. לדוגמה: --params='{"param":"param_value"}'. אלה הפרמטרים להעברה מ-Amazon S3:

  • ‫destination_table_name_template: חובה. שם טבלת היעד.

  • ‫data_path: חובה. ה-URI של Amazon S3, בפורמט הבא:

    s3://mybucket/myfolder/...

    מזהי URI תומכים גם בפרמטרים.

  • ‫access_key_id: חובה. המזהה של מפתח הגישה.

  • ‫secret_access_key: חובה. מפתח הגישה הסודי שלכם.

  • ‫file_format: אופציונלי. מציין את סוג הקבצים שרוצים להעביר: CSV,‏ JSON,‏ AVRO,‏ PARQUET או ORC. ערך ברירת המחדל הוא CSV.

  • ‫write_disposition: אופציונלי. ‫WRITE_APPEND יעביר רק את הקבצים ששונו מאז ההפעלה המוצלחת הקודמת. WRITE_TRUNCATE תעביר את כל הקבצים התואמים, כולל קבצים שהועברו בהפעלה קודמת. ערך ברירת המחדל הוא WRITE_APPEND.

  • ‫max_bad_records: אופציונלי. מספר הרשומות הפגומות המותר. ערך ברירת המחדל הוא 0.

  • ‫decimal_target_types: אופציונלי. רשימה מופרדת בפסיקים של סוגי נתונים אפשריים ב-SQL שאליהם אפשר להמיר את הערכים העשרוניים של המקור. אם לא מציינים את השדה הזה, סוג הנתונים שמוגדר כברירת מחדל הוא NUMERIC,STRING ל-ORC, ו-NUMERIC לפורמטים האחרים של הקבצים.

  • ‫ignore_unknown_values: אופציונלי, והמערכת מתעלמת ממנו אם הערך של file_format הוא לא JSON או CSV. אם להתעלם מערכים לא ידועים בנתונים.

  • ‫field_delimiter: אופציונלי, ורלוונטי רק כש-file_format הוא CSV. התו שמפריד בין השדות. ערך ברירת המחדל הוא פסיק.

  • ‫skip_leading_rows: אופציונלי, ורלוונטי רק כש-file_format הוא CSV. מציין את מספר שורות הכותרת שאתם לא רוצים לייבא. ערך ברירת המחדל הוא 0.

  • ‫allow_quoted_newlines: אופציונלי, ורלוונטי רק כש-file_format הוא CSV. מציין אם לאפשר שורות חדשות בשדות במירכאות.

  • ‫allow_jagged_rows: אופציונלי, ורלוונטי רק כש-file_format הוא CSV. מציין אם לקבל שורות שחסרות בהן עמודות אופציונליות בסוף. הערכים החסרים ימולאו בערכי NULL.

לדוגמה, הפקודה הבאה יוצרת העברת נתונים של Amazon S3 בשם My Transfer באמצעות ערך data_path של s3://mybucket/myfile/*.csv, מערך נתונים של יעד mydataset ו-file_format CSV. הדוגמה הזו כוללת ערכים שאינם ערכי ברירת המחדל של הפרמטרים האופציונליים שמשויכים לפורמט הקובץ CSV.

העברת הנתונים נוצרת בפרויקט ברירת המחדל:

bq mk --transfer_config \
--target_dataset=mydataset \
--display_name='My Transfer' \
--params='{"data_path":"s3://mybucket/myfile/*.csv",
"destination_table_name_template":"MyTable",
"file_format":"CSV",
"write_disposition":"WRITE_APPEND",
"max_bad_records":"1",
"ignore_unknown_values":"true",
"field_delimiter":"|",
"skip_leading_rows":"1",
"allow_quoted_newlines":"true",
"allow_jagged_rows":"false"}' \
--data_source=amazon_s3

אחרי שמריצים את הפקודה, מקבלים הודעה כמו זו:

[URL omitted] Please copy and paste the above URL into your web browser and follow the instructions to retrieve an authentication code.

פועלים לפי ההוראות ומדביקים את קוד האימות בשורת הפקודה.

API

משתמשים ב-method‏ projects.locations.transferConfigs.create ומספקים מופע של המשאב TransferConfig.

Java

import com.google.api.gax.rpc.ApiException;
import com.google.cloud.bigquery.datatransfer.v1.CreateTransferConfigRequest;
import com.google.cloud.bigquery.datatransfer.v1.DataTransferServiceClient;
import com.google.cloud.bigquery.datatransfer.v1.ProjectName;
import com.google.cloud.bigquery.datatransfer.v1.TransferConfig;
import com.google.protobuf.Struct;
import com.google.protobuf.Value;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

// Sample to create amazon s3 transfer config.
public class CreateAmazonS3Transfer {

  public static void main(String[] args) throws IOException {
    // TODO(developer): Replace these variables before running the sample.
    final String projectId = "MY_PROJECT_ID";
    String datasetId = "MY_DATASET_ID";
    String tableId = "MY_TABLE_ID";
    // Amazon S3 Bucket Uri with read role permission
    String sourceUri = "s3://your-bucket-name/*";
    String awsAccessKeyId = "MY_AWS_ACCESS_KEY_ID";
    String awsSecretAccessId = "AWS_SECRET_ACCESS_ID";
    String sourceFormat = "CSV";
    String fieldDelimiter = ",";
    String skipLeadingRows = "1";
    Map<String, Value> params = new HashMap<>();
    params.put(
        "destination_table_name_template", Value.newBuilder().setStringValue(tableId).build());
    params.put("data_path", Value.newBuilder().setStringValue(sourceUri).build());
    params.put("access_key_id", Value.newBuilder().setStringValue(awsAccessKeyId).build());
    params.put("secret_access_key", Value.newBuilder().setStringValue(awsSecretAccessId).build());
    params.put("source_format", Value.newBuilder().setStringValue(sourceFormat).build());
    params.put("field_delimiter", Value.newBuilder().setStringValue(fieldDelimiter).build());
    params.put("skip_leading_rows", Value.newBuilder().setStringValue(skipLeadingRows).build());
    TransferConfig transferConfig =
        TransferConfig.newBuilder()
            .setDestinationDatasetId(datasetId)
            .setDisplayName("Your Aws S3 Config Name")
            .setDataSourceId("amazon_s3")
            .setParams(Struct.newBuilder().putAllFields(params).build())
            .setSchedule("every 24 hours")
            .build();
    createAmazonS3Transfer(projectId, transferConfig);
  }

  public static void createAmazonS3Transfer(String projectId, TransferConfig transferConfig)
      throws IOException {
    try (DataTransferServiceClient client = DataTransferServiceClient.create()) {
      ProjectName parent = ProjectName.of(projectId);
      CreateTransferConfigRequest request =
          CreateTransferConfigRequest.newBuilder()
              .setParent(parent.toString())
              .setTransferConfig(transferConfig)
              .build();
      TransferConfig config = client.createTransferConfig(request);
      System.out.println("Amazon s3 transfer created successfully :" + config.getName());
    } catch (ApiException ex) {
      System.out.print("Amazon s3 transfer was not created." + ex.toString());
    }
  }
}