שימוש ב-Storage Write API ‏ (REST)

BigQuery Storage Write API (REST), שנקרא בעבר השיטה הקודמת tabledata.insertAll, מאפשר להעביר שורות בסטרימינג לטבלאות ב-BigQuery בלי להריץ משימת טעינה. הנתונים שמועברים בסטרימינג זמינים לניתוח GoogleSQL בזמן אמת ברגע ש-BigQuery מאשר את בקשת הכתיבה.

בפרויקטים חדשים, כדאי להשתמש ב-BigQuery Storage Write API (gRPC) כדי ליהנות מתמחור נמוך יותר, מסמנטיקה של מסירה פעם אחת בדיוק ומסטרימינג לטבלאות מנוהלות של Apache Iceberg. אם אתם מעבירים פרויקט קיים מ-Storage Write API ‏ (REST) ל-Storage Write API ‏ (gRPC), בחרו בזרם ברירת המחדל.

לפני שמתחילים

1. מוודאים שיש לכם גישת כתיבה למערך הנתונים שמכיל את טבלת היעד. הטבלה צריכה להיות קיימת לפני שמתחילים לכתוב בה נתונים, אלא אם משתמשים בטבלאות תבנית. מידע נוסף על טבלאות תבניות זמין במאמר יצירת טבלאות באופן אוטומטי באמצעות טבלאות תבניות.

2. בודקים את מדיניות המכסות להזרמת נתונים.

3. מוודאים שהחיוב מופעל בפרויקט ב- Google Cloud .

אי אפשר להשתמש בסטרימינג בתוכנית בחינם. אם תנסו להשתמש בסטרימינג בלי להפעיל את החיוב, תופיע השגיאה הבאה: BigQuery: Streaming insert is not allowed in the free tier.

4. להקצות תפקידים של ניהול זהויות והרשאות גישה (IAM) שנותנים למשתמשים את ההרשאות הנדרשות כדי לבצע כל משימה שמופיעה במסמך הזה.

ההרשאות הנדרשות

כדי להעביר נתונים בסטרימינג ל-BigQuery, צריך את הרשאות ה-IAM הבאות:

• bigquery.tables.updateData (מאפשרת להוסיף נתונים לטבלה)
• bigquery.tables.get (מאפשרת לקבל מטא-נתונים של טבלה)
• ‫bigquery.datasets.get (מאפשרת לקבל מטא-נתונים של מערך נתונים)
• bigquery.tables.create (חובה אם משתמשים בטבלת תבנית כדי ליצור את הטבלה באופן אוטומטי)

כל אחד מהתפקידים הבאים שמוגדרים מראש ב-IAM כולל את ההרשאות שצריך כדי להעביר נתונים בסטרימינג ל-BigQuery:

• roles/bigquery.dataEditor
• roles/bigquery.dataOwner
• roles/bigquery.admin

במאמר תפקידים והרשאות מוגדרים מראש תוכלו לקרוא מידע נוסף על תפקידים והרשאות של IAM ב-BigQuery.

העברת נתונים בסטרימינג ל-BigQuery

C#

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של C# במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery C# API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

using Google.Cloud.BigQuery.V2;

public class BigQueryTableInsertRows
{
    public void TableInsertRows(
        string projectId = "your-project-id",
        string datasetId = "your_dataset_id",
        string tableId = "your_table_id"
    )
    {
        BigQueryClient client = BigQueryClient.Create(projectId);
        BigQueryInsertRow[] rows = new BigQueryInsertRow[]
        {
            // The insert ID is optional, but can avoid duplicate data
            // when retrying inserts.
            new BigQueryInsertRow(insertId: "row1") {
                { "name", "Washington" },
                { "post_abbr", "WA" }
            },
            new BigQueryInsertRow(insertId: "row2") {
                { "name", "Colorado" },
                { "post_abbr", "CO" }
            }
        };
        client.InsertRows(datasetId, tableId, rows);
    }
}

Go

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Go במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery Go API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

import (
	"context"
	"fmt"

	"cloud.google.com/go/bigquery"
)

// Item represents a row item.
type Item struct {
	Name string
	Age  int
}

// Save implements the ValueSaver interface.
// This example disables best-effort de-duplication, which allows for higher throughput.
func (i *Item) Save() (map[string]bigquery.Value, string, error) {
	return map[string]bigquery.Value{
		"full_name": i.Name,
		"age":       i.Age,
	}, bigquery.NoDedupeID, nil
}

// insertRows demonstrates inserting data into a table using the streaming insert mechanism.
func insertRows(projectID, datasetID, tableID string) error {
	// projectID := "my-project-id"
	// datasetID := "mydataset"
	// tableID := "mytable"
	ctx := context.Background()
	client, err := bigquery.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("bigquery.NewClient: %w", err)
	}
	defer client.Close()

	inserter := client.Dataset(datasetID).Table(tableID).Inserter()
	items := []*Item{
		// Item implements the ValueSaver interface.
		{Name: "Phred Phlyntstone", Age: 32},
		{Name: "Wylma Phlyntstone", Age: 29},
	}
	if err := inserter.Put(ctx, items); err != nil {
		return err
	}
	return nil
}

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery Java API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryError;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.InsertAllRequest;
import com.google.cloud.bigquery.InsertAllResponse;
import com.google.cloud.bigquery.TableId;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Sample to inserting rows into a table without running a load job.
public class TableInsertRows {

  public static void main(String[] args) {
    // TODO(developer): Replace these variables before running the sample.
    String datasetName = "MY_DATASET_NAME";
    String tableName = "MY_TABLE_NAME";
    // Create a row to insert
    Map<String, Object> rowContent = new HashMap<>();
    rowContent.put("booleanField", true);
    rowContent.put("numericField", "3.14");
    // TODO(developer): Replace the row id with a unique value for each row.
    String rowId = "ROW_ID";
    tableInsertRows(datasetName, tableName, rowId, rowContent);
  }

  public static void tableInsertRows(
      String datasetName, String tableName, String rowId, Map<String, Object> rowContent) {
    try {
      // Initialize client that will be used to send requests. This client only needs to be created
      // once, and can be reused for multiple requests.
      BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();

      // Get table
      TableId tableId = TableId.of(datasetName, tableName);

      // Inserts rowContent into datasetName:tableId.
      InsertAllResponse response =
          bigquery.insertAll(
              InsertAllRequest.newBuilder(tableId)
                  // More rows can be added in the same RPC by invoking .addRow() on the builder.
                  // You can omit the unique row ids to disable de-duplication.
                  .addRow(rowId, rowContent)
                  .build());

      if (response.hasErrors()) {
        // If any of the insertions failed, this lets you inspect the errors
        for (Map.Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) {
          System.out.println("Response error: \n" + entry.getValue());
        }
      }
      System.out.println("Rows successfully inserted into table");
    } catch (BigQueryException e) {
      System.out.println("Insert operation not performed \n" + e.toString());
    }
  }
}

Node.js

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Node.js במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery Node.js API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

// Import the Google Cloud client library
const {BigQuery} = require('@google-cloud/bigquery');
const bigquery = new BigQuery();

async function insertRowsAsStream() {
  // Inserts the JSON objects into my_dataset:my_table.

  /**
   * TODO(developer): Uncomment the following lines before running the sample.
   */
  // const datasetId = 'my_dataset';
  // const tableId = 'my_table';
  const rows = [
    {name: 'Tom', age: 30},
    {name: 'Jane', age: 32},
  ];

  // Insert data into a table
  await bigquery.dataset(datasetId).table(tableId).insert(rows);
  console.log(`Inserted ${rows.length} rows`);
}

PHP

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של PHP במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery PHP API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

use Google\Cloud\BigQuery\BigQueryClient;

/**
 * Stream data into bigquery
 *
 * @param string $projectId The project Id of your Google Cloud Project.
 * @param string $datasetId The BigQuery dataset ID.
 * @param string $tableId The BigQuery table ID.
 * @param string $data Json encoded data For eg,
 *    $data = json_encode([
 *       "field1" => "value1",
 *       "field2" => "value2",
 *    ]);
 */
function stream_row(
    string $projectId,
    string $datasetId,
    string $tableId,
    string $data
): void {
    // instantiate the bigquery table service
    $bigQuery = new BigQueryClient([
      'projectId' => $projectId,
    ]);
    $dataset = $bigQuery->dataset($datasetId);
    $table = $dataset->table($tableId);

    $data = json_decode($data, true);
    $insertResponse = $table->insertRows([
      ['data' => $data],
      // additional rows can go here
    ]);
    if ($insertResponse->isSuccessful()) {
        print('Data streamed into BigQuery successfully' . PHP_EOL);
    } else {
        foreach ($insertResponse->failedRows() as $row) {
            foreach ($row['errors'] as $error) {
                printf('%s: %s' . PHP_EOL, $error['reason'], $error['message']);
            }
        }
    }
}

Python

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Python במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery Python API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

from google.cloud import bigquery

# Construct a BigQuery client object.
client = bigquery.Client()

# TODO(developer): Set table_id to the ID of table to append to.
# table_id = "your-project.your_dataset.your_table"

rows_to_insert = [
    {"full_name": "Phred Phlyntstone", "age": 32},
    {"full_name": "Wylma Phlyntstone", "age": 29},
]

errors = client.insert_rows_json(table_id, rows_to_insert)  # Make an API request.
if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

Ruby

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Ruby במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery Ruby API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

require "google/cloud/bigquery"

def table_insert_rows dataset_id = "your_dataset_id", table_id = "your_table_id"
  bigquery = Google::Cloud::Bigquery.new
  dataset  = bigquery.dataset dataset_id
  table    = dataset.table table_id

  row_data = [
    { name: "Alice", value: 5  },
    { name: "Bob",   value: 10 }
  ]
  response = table.insert row_data

  if response.success?
    puts "Inserted rows successfully"
  else
    puts "Failed to insert #{response.error_rows.count} rows"
  end
end

הערה: כדי לציין ערך של NUMERIC או BIGNUMERIC בשורה, צריך להקיף את הערך במירכאות כפולות, למשל "big_numeric_col":"0.123456789123".

אין צורך לאכלס את השדה insertID כשמוסיפים שורות. בדוגמה הבאה מוסבר איך להימנע משליחת insertID לכל שורה בזמן סטרימינג.

Java

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Java במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery Java API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryError;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.InsertAllRequest;
import com.google.cloud.bigquery.InsertAllResponse;
import com.google.cloud.bigquery.TableId;
import com.google.common.collect.ImmutableList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Sample to insert rows without row ids in a table
public class TableInsertRowsWithoutRowIds {

  public static void main(String[] args) {
    // TODO(developer): Replace these variables before running the sample.
    String datasetName = "MY_DATASET_NAME";
    String tableName = "MY_TABLE_NAME";
    tableInsertRowsWithoutRowIds(datasetName, tableName);
  }

  public static void tableInsertRowsWithoutRowIds(String datasetName, String tableName) {
    try {
      // Initialize client that will be used to send requests. This client only needs to be created
      // once, and can be reused for multiple requests.
      final BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
      // Create rows to insert
      Map<String, Object> rowContent1 = new HashMap<>();
      rowContent1.put("stringField", "Phred Phlyntstone");
      rowContent1.put("numericField", 32);
      Map<String, Object> rowContent2 = new HashMap<>();
      rowContent2.put("stringField", "Wylma Phlyntstone");
      rowContent2.put("numericField", 29);
      InsertAllResponse response =
          bigquery.insertAll(
              InsertAllRequest.newBuilder(TableId.of(datasetName, tableName))
                  // No row ids disable de-duplication, and also disable the retries in the Java
                  // library.
                  .setRows(
                      ImmutableList.of(
                          InsertAllRequest.RowToInsert.of(rowContent1),
                          InsertAllRequest.RowToInsert.of(rowContent2)))
                  .build());

      if (response.hasErrors()) {
        // If any of the insertions failed, this lets you inspect the errors
        for (Map.Entry<Long, List<BigQueryError>> entry : response.getInsertErrors().entrySet()) {
          System.out.println("Response error: \n" + entry.getValue());
        }
      }
      System.out.println("Rows successfully inserted into table without row ids");
    } catch (BigQueryException e) {
      System.out.println("Insert operation not performed \n" + e.toString());
    }
  }
}

Python

לפני שמנסים את הדוגמה הזו, צריך לפעול לפי הוראות ההגדרה של Python במדריך למתחילים של BigQuery באמצעות ספריות לקוח. למידע נוסף, קראו את מאמרי העזרה של BigQuery Python API.

כדי לבצע אימות ב-BigQuery, צריך להגדיר את Application Default Credentials. מידע נוסף זמין במאמר הגדרת אימות לספריות לקוח.

from google.cloud import bigquery

# Construct a BigQuery client object.
client = bigquery.Client()

# TODO(developer): Set table_id to the ID of table to append to.
# table_id = "your-project.your_dataset.your_table"

rows_to_insert = [
    {"full_name": "Phred Phlyntstone", "age": 32},
    {"full_name": "Wylma Phlyntstone", "age": 29},
]

errors = client.insert_rows_json(
    table_id, rows_to_insert, row_ids=[None] * len(rows_to_insert)
)  # Make an API request.
if errors == []:
    print("New rows have been added.")
else:
    print("Encountered errors while inserting rows: {}".format(errors))

שליחת נתונים של תאריך ושעה

בשדות של תאריך ושעה, צריך להזין את הנתונים בפורמט הבא ב-Storage Write API ‏ (REST):

סוג	פורמט
DATE	מחרוזת בפורמט "YYYY-MM-DD"
DATETIME	מחרוזת בפורמט "YYYY-MM-DD [HH:MM:SS]"
TIME	מחרוזת בפורמט "HH:MM:SS"
TIMESTAMP	מספר השניות מאז 1970-01-01 (ראשית זמן יוניקס), או מחרוזת בפורמט "YYYY-MM-DD HH:MM[:SS]"

שליחת נתוני טווח

בשדות מסוג RANGE<T>, צריך להגדיר את הפורמט של הנתונים ב-Storage Write API ‏ (REST) כאובייקט JSON עם שני שדות, start ו-end. ערכים חסרים או ערכי NULL בשדות start ו-end מייצגים גבולות לא מוגדרים. השדות האלה חייבים להיות באותו פורמט JSON נתמך מסוג T, כאשר T יכול להיות אחד מהערכים DATE,‏ DATETIME ו-TIMESTAMP.

בדוגמה הבאה, השדה f_range_date מייצג עמודה מסוג RANGE<DATE> בטבלה. שורה מוכנסת לעמודה הזו באמצעות Storage Write API ‏ (REST).

{
    "f_range_date": {
        "start": "1970-01-02",
        "end": null
    }
}

זמינות הנתונים של מקור הנתונים

הנתונים זמינים לניתוח בזמן אמת באמצעות שאילתות GoogleSQL מיד אחרי ש-BigQuery מאשרת בקשה של Storage Write API ‏ (REST). כששולחים שאילתות לנתונים במאגר הנתונים שנכנסים בסטרימינג, לא מחויבים על בייטים שעובדו ממאגר הנתונים שנכנסים בסטרימינג אם משתמשים בתמחור על פי דרישה של משאבי המחשוב. אם אתם משתמשים בתמחור מבוסס-קיבולת, ההזמנות שלכם צורכות משבצות לעיבוד נתונים במאגר נתונים (buffer) של סטרימינג.

לשורות שהועברו לאחרונה בסטרימינג לטבלה מחולקת למחיצות (Partitions) לפי זמני כתיבת הנתונים, יש באופן זמני ערך NULL בעמודה המדומה _PARTITIONTIME. בשורות כאלה, BigQuery מקצה ברקע את הערך הסופי שאינו NULL של העמודה PARTITIONTIME, בדרך כלל תוך כמה דקות. במקרים נדירים, התהליך יכול להימשך עד 90 דקות.

יכול להיות שחלק מהשורות שנוספו לאחרונה בסטרימינג לא יהיו זמינות להעתקה של הטבלה, בדרך כלל למשך כמה דקות. במקרים נדירים, התהליך יכול להימשך עד 90 דקות. כדי לבדוק אם הנתונים זמינים להעתקת טבלה, בודקים את התגובה של tables.get בקטע שנקרא streamingBuffer. אם הקטע streamingBuffer לא מופיע, אפשר להעתיק את הנתונים. אפשר גם להשתמש בשדה streamingBuffer.oldestEntryTime כדי לזהות את הגיל של הרשומות במאגר הנתונים הזמני של הסטרימינג.

ביטול כפילויות במידת האפשר

כשמציינים insertId לשורה שנוספה, BigQuery משתמש במזהה הזה כדי לתמוך בהסרת כפילויות כמיטב היכולת, למשך עד דקה אחת. כלומר, אם תעבירו בסטרימינג את אותה שורה עם אותו insertId יותר מפעם אחת באותה תקופה לאותה טבלה, יכול להיות ש-BigQuery יסיר את המופעים הכפולים של השורה הזו וישמור רק אחד מהם.

המערכת מצפה ששורות עם insertId זהים יהיו גם הן זהות. אם לשתי שורות יש ערכי insertId זהים, לא ניתן לקבוע מראש איזו שורה תישמר ב-BigQuery.

ביטול כפילויות מיועד בדרך כלל לתרחישים של ניסיון חוזר במערכת מבוזרת, שבהם אין דרך לקבוע את המצב של הזנת זרם נתונים בתנאי שגיאה מסוימים, כמו שגיאות בחיבור לרשת בין המערכת שלכם לבין BigQuery או שגיאות פנימיות ב-BigQuery. אם אתם מנסים שוב להוסיף, השתמשו באותו insertId לאותה קבוצת שורות כדי ש-BigQuery יוכל לנסות לבטל את הכפילויות בנתונים. למידע נוסף, קראו את המאמר פתרון בעיות בהזנת זרם נתונים.

הסרת הכפילויות שמוצעת על ידי BigQuery היא כמיטב היכולת, ואין להסתמך עליה כמנגנון להבטחת היעדר כפילויות בנתונים. בנוסף, יכול להיות ש-BigQuery יפחית את האיכות של הסרת הכפילויות כמיטב היכולת בכל שלב, כדי להבטיח אמינות וזמינות גבוהות יותר של הנתונים.

אם יש לכם דרישות מחמירות לגבי ביטול כפילויות בנתונים, Google Cloud Datastore הוא שירות חלופי שתומך בטרנזקציות.

השבתת ביטול כפילויות בהקדם האפשרי

כדי להשבית את מניעת הכפילויות, אל תמלאו את השדה insertId בכל שורה שאתם מוסיפים. זו הדרך המומלצת להוסיף נתונים.

‫Apache Beam ו-Dataflow

כדי להשבית את ביטול הכפילויות במידת האפשר כשמשתמשים במחבר הקלט/פלט של BigQuery ב-Apache Beam ל-Java, משתמשים בmethod‏ ignoreInsertIds().

הסרה ידנית של כפילויות

כדי לוודא שאין שורות כפולות אחרי שמסיימים את הסטרימינג, צריך לבצע את התהליך הידני הבא:

1. מוסיפים את insertId כעמודה בסכימת הטבלה וכוללים את הערך insertId בנתונים של כל שורה.
2. אחרי שהסטרימינג נפסק, מריצים את השאילתה הבאה כדי לבדוק אם יש כפילויות:

   #standardSQL
   SELECT
     MAX(count) FROM(
     SELECT
       ID_COLUMN,
       count(*) as count
     FROM
       `TABLE_NAME`
     GROUP BY
       ID_COLUMN)

   אם התוצאה גדולה מ-1, יש כפילויות.
3. כדי להסיר כפילויות, מריצים את השאילתה הבאה. מציינים טבלת יעד, מאפשרים תוצאות גדולות ומשביתים את שיטוח התוצאות.

   #standardSQL
   SELECT
     * EXCEPT(row_number)
   FROM (
     SELECT
       *,
       ROW_NUMBER()
             OVER (PARTITION BY ID_COLUMN) row_number
     FROM
       `TABLE_NAME`)
   WHERE
     row_number = 1

הערות לגבי השאילתה להסרת כפילויות:

• האסטרטגיה הבטוחה יותר לשאילתה להסרת כפילויות היא לטרגט טבלה חדשה. אפשר גם לטרגט את טבלת המקור עם מאפיין הכתיבה WRITE_TRUNCATE.
• השאילתה להסרת כפילויות מוסיפה עמודת row_number עם הערך 1 לסוף הסכימה של הטבלה. השאילתה משתמשת בהצהרה SELECT * EXCEPT מ-GoogleSQL כדי להחריג את העמודה row_number מטבלת היעד. הקידומת #standardSQL מפעילה את GoogleSQL בשאילתה הזו. לחלופין, אפשר לבחור שמות ספציפיים של עמודות כדי להשמיט את העמודה הזו.
• כדי לשלוח שאילתות על נתונים בזמן אמת ללא כפילויות, אתם יכולים גם ליצור תצוגה של הטבלה באמצעות השאילתה להסרת כפילויות. חשוב לזכור שעלויות השאילתות בתצוגה מחושבות על סמך העמודות שנבחרו בתצוגה, מה שיכול להוביל לגודל גדול של בייטים שנסרקו.

העברה בסטרימינג לטבלאות שמחולקות למחיצות לפי זמן

כשמעבירים נתונים בסטרימינג לטבלה שמחולקת למחיצות לפי זמן, לכל מחיצה יש מאגר נתונים זמני לסטרימינג. מאגר הנתונים הזמני של הסטרימינג נשמר כשמבצעים משימת טעינה, שאילתה או העתקה שמחליפה מחיצה על ידי הגדרת המאפיין writeDisposition לערך WRITE_TRUNCATE. כדי להסיר את מאגר הנתונים הזמני של הסטרימינג, צריך לוודא שהוא ריק באמצעות קריאה ל-tables.get במחיצה.

חלוקה למחיצות לפי זמן הטמעה

כשמעבירים נתונים בסטרימינג לטבלה שמחולקת למחיצות לפי זמני הטמעת הנתונים, BigQuery מסיק את המחיצה של היעד לפי זמן ה-UTC הנוכחי.

נתונים חדשים שמתקבלים מועברים באופן זמני למחיצה __UNPARTITIONED__ בזמן שהם נמצאים במאגר הנתונים (buffer) להעברה בסטרימינג. כשיש מספיק נתונים לא מחולקים למחיצות, BigQuery מחלק את הנתונים למחיצה הנכונה. עם זאת, אין הסכם SLA לגבי משך הזמן שלוקח להעביר נתונים מהמחיצה __UNPARTITIONED__. כדי להחריג נתונים במאגר הנתונים הזמני משאילתה, אפשר לסנן את הערכים של NULL מהמחיצה __UNPARTITIONED__ באמצעות אחת מהעמודות המדומה (_PARTITIONTIME או _PARTITIONDATE, בהתאם לסוג הנתונים המועדף).

אם אתם מעבירים נתונים בסטרימינג לטבלה שמחולקת למחיצות לפי יום, אתם יכולים לשנות את הסקת התאריך על ידי הוספת מעצב מחיצה כחלק מ-Storage Write API ‏ (REST). כוללים את ה-decorator בפרמטר tableId. לדוגמה, אפשר להעביר בסטרימינג למחיצה שתואמת ל-2021-03-01 בטבלה table1 באמצעות מעטר המחיצה:

table1$20210301

כשמבצעים סטרימינג באמצעות מעטר מחיצות, אפשר להעביר נתונים למחיצות בטווח של 31 ימים בעבר ו-16 ימים בעתיד, ביחס לתאריך הנוכחי, על סמך זמן UTC הנוכחי. כדי לכתוב למחיצות של תאריכים מחוץ לגבולות המותרים האלה, צריך להשתמש במקום זאת במשימת טעינה או במשימת שאילתה, כפי שמתואר בקטע הוספה לנתוני טבלה מחולקת למחיצות והחלפתם.

אפשר להשתמש ב-partition decorator רק בטבלאות שמחולקות למחיצות לפי יום. הוא לא נתמך בטבלאות שמחולקות למחיצות לפי שעה, חודש או שנה.

כדי לבצע בדיקה, אפשר להשתמש בפקודת ה-CLI bq insert של כלי שורת הפקודה של BigQuery. לדוגמה, הפקודה הבאה מעבירה בסטרימינג שורה אחת למחיצה של התאריך 1 בינואר 2017 ($20170101) בטבלה מחולקת למחיצות בשם mydataset.mytable:

echo '{"a":1, "b":2}' | bq insert 'mydataset.mytable$20170101'

זהירות: הפקודה bq insert מיועדת לבדיקה בלבד.

חלוקה למחיצות לפי עמודת יחידת זמן

אפשר להעביר נתונים בסטרימינג לטבלה שמחולקת למחיצות לפי עמודה מסוג DATE, DATETIME או TIMESTAMP, בטווח שבין 10 שנים בעבר לשנה אחת בעתיד. נתונים מחוץ לטווח הזה נדחים.

כשהנתונים מועברים בסטרימינג, הם ממוקמים בהתחלה במחיצה __UNPARTITIONED__. כשיש מספיק נתונים לא מחולקים למחיצות, BigQuery מחלק אותם מחדש באופן אוטומטי ומציב אותם במחיצה המתאימה. עם זאת, אין הסכם SLA לגבי משך הזמן שלוקח להעביר נתונים מהמחיצה __UNPARTITIONED__.

• הערה: מחיצות יומיות מעובדות באופן שונה ממחיצות לפי שעה, חודש ושנה. רק נתונים מחוץ לטווח התאריכים (מ-7 הימים האחרונים ועד 3 הימים הבאים) מחולצים למחיצה UNPARTITIONED, וממתינים לחלוקה מחדש למחיצות. לעומת זאת, בטבלה שמחולקת למחיצות לפי שעה, הנתונים תמיד מחולצים למחיצה UNPARTITIONED, ואחר כך מחולקים מחדש למחיצות.

יצירת טבלאות באופן אוטומטי באמצעות תבניות של טבלאות

טבלאות תבנית מספקות מנגנון לפיצול טבלה לוגית להרבה טבלאות קטנות יותר, כדי ליצור קבוצות קטנות יותר של נתונים (לדוגמה, לפי מזהה משתמש). לטבלאות של תבניות יש כמה מגבלות, שמתוארות בהמשך הקטע הזה. במקום זאת, מומלץ להשתמש בטבלאות מחולקות למחיצות ובטבלאות מקובצות לאשכולות כדי להשיג את ההתנהגות הזו.

כדי להשתמש בטבלת תבנית דרך BigQuery API, מוסיפים פרמטר templateSuffix לבקשת Storage Write API (REST). בכלי שורת הפקודה של BigQuery‏ (bq), מוסיפים את הדגל template_suffix לפקודה insert. אם BigQuery מזהה פרמטר templateSuffix או דגל template_suffix, הוא מתייחס לטבלת היעד כאל תבנית בסיס. היא יוצרת טבלה חדשה עם אותה סכימה כמו של טבלת היעד, והשם שלה כולל את הסיומת שצוינה:

<targeted_table_name> + <templateSuffix>

שימוש בטבלת תבנית חוסך את הצורך ליצור כל טבלה בנפרד ולציין את הסכימה של כל טבלה. אתם צריכים ליצור רק תבנית אחת ולספק סיומות שונות, כדי ש-BigQuery יוכל ליצור בשבילכם את הטבלאות החדשות. ‫BigQuery ממקם את הטבלאות באותו פרויקט ובאותו מערך נתונים.

טבלאות שנוצרות באמצעות טבלאות תבנית זמינות בדרך כלל תוך כמה שניות. במקרים נדירים, יכול להיות שייקח יותר זמן עד שהם יהיו זמינים.

שינוי הסכימה של טבלת התבנית

אם משנים סכימה של טבלת תבנית, כל הטבלאות שנוצרות לאחר מכן משתמשות בסכימה המעודכנת. טבלאות שנוצרו בעבר לא מושפעות, אלא אם בטבלה הקיימת עדיין יש מאגר נתונים זמני (buffer) בסטרימינג.

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

אחרי שמשנים סכימה של טבלת תבנית, צריך להמתין עד שהשינויים יופצו לפני שמנסים להוסיף נתונים חדשים או לשלוח שאילתות לטבלאות שנוצרו. בקשות להוספת שדות חדשים אמורות להצליח תוך כמה דקות. יכול להיות שתצטרכו להמתין עד 90 דקות לפני שתנסו לשלוח שאילתות לשדות החדשים.

אם רוצים לשנות את הסכימה של טבלה שנוצרה, צריך להמתין עד שהסטרימינג דרך טבלת התבנית יסתיים, ועד שהקטע של נתוני הסטרימינג של הטבלה שנוצרה לא יופיע בתגובה של tables.get(), מה שמצביע על כך שאין נתונים במאגר הנתונים הזמני בטבלה.

טבלאות מחולקות למחיצות וטבלאות מקובצות באשכולות לא מושפעות מהמגבלות האלה, והן המנגנון המומלץ.

פרטי טבלת התבניות

ערך הסיומת של התבנית

הערך של templateSuffix (או --template_suffix) יכול להכיל רק אותיות (a-z,‏ A-Z), מספרים (0-9) או קווים תחתונים (_). האורך המקסימלי המשולב של שם הטבלה והסיומת של הטבלה הוא 1,024 תווים.

מכסה

טבלאות תבניות כפופות למגבלות של מכסת סטרימינג. בפרויקט אפשר ליצור עד 10 טבלאות בשנייה באמצעות טבלאות תבנית, בדומה ל-API של tables.insert. המכסה הזו חלה רק על טבלאות שנוצרות, ולא על טבלאות שמשתנות.

אם האפליקציה צריכה ליצור יותר מ-10 טבלאות בשנייה, מומלץ להשתמש בטבלאות מסודרות באשכולות. לדוגמה, אפשר להזין את מזהה הטבלה עם עוצמה גבוהה בעמודת המפתח של טבלה אחת שמסודרת באשכולות.

זמן חיים (TTL)

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

ביטול כפילויות

ביטול כפילויות מתרחש רק בין הפניות אחידות לטבלת יעד. לדוגמה, אם אתם מעבירים בסטרימינג בו-זמנית לטבלה שנוצרה באמצעות טבלאות תבנית ופקודה רגילה של Storage Write API (REST), לא מתבצע ביטול כפילויות בין שורות שהוכנסו על ידי טבלאות תבנית לבין פקודה רגילה של Storage Write API (REST).

תצוגות

טבלת התבנית והטבלאות שנוצרו לא יכולות להיות תצוגות.

פתרון בעיות בהזנת זרם נתונים

בקטעים הבאים מוסבר איך לפתור שגיאות שמתרחשות כשמעבירים נתונים בסטרימינג ל-BigQuery באמצעות Storage Write API ‏ (REST). איך פותרים שגיאות מכסה בהזנת זרם נתונים

קודי תגובת HTTP של שגיאות

אם אתם מקבלים קוד תגובת HTTP של כישלון, כמו שגיאה בחיבור לרשת, אי אפשר לדעת אם הזנת זרם הנתונים הצליחה. אם תנסו לשלוח מחדש את הבקשה, יכול להיות שיופיעו שורות כפולות בטבלה. כדי להגן על הטבלה מפני שכפול, מגדירים את המאפיין insertId כששולחים את הבקשה. ‫BigQuery משתמש במאפיין insertId כדי להסיר כפילויות.

אם מתקבלת שגיאת הרשאה, שגיאה של שם טבלה לא חוקי או שגיאה של חריגה מהמכסה, לא מוכנסות שורות וכל הבקשה נכשלת.

קודי תגובת HTTP שמציינים הצלחה

גם אם קיבלתם קוד תגובת HTTP שמציין הצלחה, אתם צריכים לבדוק את המאפיין insertErrors של התגובה כדי לקבוע אם הוספת השורות הצליחה, כי יכול להיות ש-BigQuery הצליח להוסיף רק חלק מהשורות. יכול להיות שתיתקלו באחד מהתרחישים הבאים:

• כל השורות נוספו בהצלחה: אם המאפיין insertErrors הוא רשימה ריקה, כל השורות נוספו בהצלחה.
• חלק מהשורות נוספו בהצלחה: חוץ ממקרים שבהם יש חוסר התאמה בסכימה באחת מהשורות, השורות שצוינו במאפיין insertErrors לא נוספות, וכל שאר השורות נוספות בהצלחה. המאפיין errors מכיל מידע מפורט על הסיבה לכך שכל שורה לא הצליחה. המאפיין index מציין את אינדקס השורה של הבקשה שהשגיאה חלה עליה, החל מ-0.
• לא הוכנסו שורות: אם BigQuery נתקל בחוסר התאמה לסכימה בשורות מסוימות בבקשה, אף אחת מהשורות לא מוכנסת ומוחזרת רשומה של insertErrors לכל שורה, גם לשורות שלא הייתה בהן חוסר התאמה לסכימה. שורות שלא הייתה בהן אי-התאמה לסכימה כוללות שגיאה שבה המאפיין reason מוגדר כ-stopped, ואפשר לשלוח אותן מחדש כמו שהן. שורות שנכשלו כוללות מידע מפורט על אי-ההתאמה לסכימה. למידע על סוגי מאגרי האחסון לפרוטוקולים שנתמכים בכל סוג נתונים של BigQuery, קראו את המאמר סוגי הנתונים הנתמכים של מאגרי האחסון לפרוטוקולים ו-Arrow.

שגיאות במטא-נתונים של הזנת זרם נתונים

ממשק ה-API של BigQuery להעברת נתונים בסטרימינג מיועד לשיעורי הוספה גבוהים, ולכן שינויים במטא-נתונים של הטבלה הבסיסית הם במודל עקביות הדרגתי כשמקיימים אינטראקציה עם מערכת הסטרימינג. ברוב המקרים, שינויים במטא-נתונים מתפשטים תוך כמה דקות, אבל במהלך התקופה הזו התגובות מה-API עשויות לשקף את המצב הלא עקבי של הטבלה.

לדוגמה:

• שינויים בסכימה: שינוי הסכימה של טבלה שקיבלה לאחרונה הוספות של נתונים בסטרימינג עלול לגרום לתגובות עם שגיאות של אי-התאמה בסכימה, כי יכול להיות שמערכת הסטרימינג לא תזהה מיד את השינוי בסכימה.
• יצירה או מחיקה של טבלה: אם תנסו להעביר נתונים בסטרימינג לטבלה שלא קיימת, תקבלו וריאציה של התגובה notFound. יכול להיות שטבלה שנוצרה בתגובה לא תזוהה מיד על ידי הזנות של זרמי נתונים לאחר מכן. באופן דומה, אם מוחקים או יוצרים מחדש טבלה, יכול להיות שיהיה פרק זמן שבו הזנת זרם הנתונים תועבר לטבלה הישנה. יכול להיות שהזנות של זרם נתונים לא יופיעו בטבלה החדשה.
• חיתוך טבלה: חיתוך הנתונים של טבלה (באמצעות משימת שאילתה שמשתמשת בערך writeDisposition של WRITE_TRUNCATE) יכול לגרום באופן דומה להשמטת הוספות עוקבות במהלך תקופת העקביות.

נתונים חסרים או לא זמינים

הזנות של זרמי נתונים נמצאות באופן זמני באחסון שעבר אופטימיזציה לכתיבה, שמאפייני הזמינות שלו שונים מאלה של האחסון המנוהל. פעולות מסוימות ב-BigQuery לא פועלות עם אחסון באופטימיזציה לכתיבה, כמו משימות להעתקת טבלאות ושיטות API כמו tabledata.list. נתונים עדכניים בסטרימינג לא מופיעים בטבלת היעד או בפלט.

שגיאות שקשורות למכסה של הזנת זרם נתונים

בקטע הזה מפורטים טיפים לפתרון בעיות שקשורות לשגיאות במכסות של הזרמת נתונים ל-BigQuery.

באזורים מסוימים, המכסה של הזנת זרם נתונים גדולה יותר אם לא מאכלסים את השדה insertId בכל שורה. מידע נוסף על מכסות להזנת זרם נתונים זמין במאמר הזנת זרם נתונים. השגיאות שקשורות למכסות בסטרימינג של BigQuery תלויות בנוכחות או בהיעדר של insertId.

הודעת שגיאה

אם השדה insertId ריק, יכול להיות שתופיע שגיאת המכסה הבאה:

מגבלת המכסה	הודעת השגיאה
בייטים לשנייה לכל פרויקט	הישות עם gaia_id: ‏GAIA_ID, פרויקט: PROJECT_ID באזור: REGION חרגה מהמכסה של בייטים להוספה בשנייה.

אם השדה insertId מאוכלס, יכולות להופיע שגיאות המכסה הבאות:

מגבלת המכסה	הודעת השגיאה
שורות לשנייה לכל פרויקט	הפרויקט PROJECT_ID ב-REGION חרג מהמכסה של שורות להזנת זרם נתונים לשנייה.
שורות לשנייה לכל טבלה	הטבלה TABLE_ID חרגה מהמכסה של שורות לשנייה בהזנת זרם נתונים.
בייטים לשנייה לכל טבלה	הטבלה TABLE_ID חרגה מהמכסה של בייטים לשנייה בהזנת זרם נתונים.

מטרת השדה insertId היא למנוע כפילויות בשורות שנוספו. אם כמה הוספות עם אותו insertId מגיעות בטווח של כמה דקות, BigQuery כותב גרסה אחת של הרשומה. עם זאת, אין הבטחה שהביטול האוטומטי של הכפילויות יתבצע. כדי להשיג קצב העברה מקסימלי, מומלץ לא לכלול את insertId ולהשתמש במקום זאת בביטול כפילויות ידני. מידע נוסף זמין במאמר שמירה על עקביות הנתונים.

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

האבחון

משתמשים בתצוגות STREAMING_TIMELINE_BY_* כדי לנתח את תנועת הנתונים בסטרימינג. התצוגות האלה צוברות נתונים סטטיסטיים של סטרימינג במרווחים של דקה, ומקובצות לפי error_code. שגיאות שקשורות למכסות מופיעות בתוצאות עם הערך error_code ששווה ל-RATE_LIMIT_EXCEEDED או ל-QUOTA_EXCEEDED.

בהתאם למגבלת המכסה הספציפית שהגעתם אליה, עיינו ב-total_rows או ב-total_input_bytes. אם השגיאה היא מכסה ברמת הטבלה, מסננים לפי table_id.

לדוגמה, השאילתה הבאה מציגה את סך הבייטים שנקלטו לדקה ואת המספר הכולל של שגיאות המכסה:

SELECT
 start_timestamp,
 error_code,
 SUM(total_input_bytes) as sum_input_bytes,
 SUM(IF(error_code IN ('QUOTA_EXCEEDED', 'RATE_LIMIT_EXCEEDED'),
     total_requests, 0)) AS quota_error
FROM
 `region-REGION_NAME`.INFORMATION_SCHEMA.STREAMING_TIMELINE_BY_PROJECT
WHERE
  start_timestamp > TIMESTAMP_SUB(CURRENT_TIMESTAMP, INTERVAL 1 DAY)
GROUP BY
 start_timestamp,
 error_code
ORDER BY 1 DESC

רזולוציה

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

• אם אתם משתמשים בשדה insertId לביטול כפילויות, והפרויקט שלכם נמצא באזור שתומך במכסת הסטרימינג הגבוהה יותר, מומלץ להסיר את השדה insertId. יכול להיות שהפתרון הזה ידרוש כמה שלבים נוספים כדי לבטל את הכפילויות בנתונים באופן ידני. מידע נוסף זמין במאמר הסרה ידנית של כפילויות.

• אם אתם לא משתמשים ב-insertId, או אם אי אפשר להסיר אותו, עקבו אחרי תנועת הסטרימינג במשך 24 שעות ונתחו את שגיאות המכסה:

  • אם רוב השגיאות הן מסוג RATE_LIMIT_EXCEEDED ולא מסוג QUOTA_EXCEEDED, והתנועה הכוללת היא פחות מ-80% מהמכסה, כנראה שהשגיאות מצביעות על עליות חדות זמניות. כדי לטפל בשגיאות האלה, אפשר לנסות שוב את הפעולה באמצעות השהיה מעריכית לפני ניסיון חוזר (exponential backoff) בין הניסיונות החוזרים.

  • אם אתם משתמשים במשימת Dataflow כדי להוסיף נתונים, כדאי להשתמש במשימות טעינה במקום בהזנת זרם נתונים. מידע נוסף זמין במאמר הגדרת שיטת ההוספה. אם אתם משתמשים ב-Dataflow עם מחבר קלט/פלט בהתאמה אישית, כדאי להשתמש במחבר קלט/פלט מובנה. מידע נוסף מופיע במאמר דפוסי קלט/פלט בהתאמה אישית.

  • אם אתם רואים שגיאות QUOTA_EXCEEDED או שהתנועה הכוללת חורגת באופן עקבי מ-80% מהמכסה, שלחו בקשה להגדלת המכסה. איך שולחים בקשה לשינוי המכסות

  • כדאי גם לשקול להחליף את הזנת זרם הנתונים ב-Storage Write API החדש יותר, שיש לו תפוקה גבוהה יותר, מחיר נמוך יותר והרבה תכונות שימושיות.