שימוש ב-Legacy streaming API

במאמר הזה מוסבר איך להזרים נתונים ל-BigQuery באמצעות השיטה הקודמת של tabledata.insertAll.

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

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

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

2. כדאי לעיין במדיניות בנושא מכסות לנתונים שמוזרמים.

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

הסטרימינג לא זמין דרך התוכנית ללא תשלום. אם תנסו להשתמש בסטרימינג בלי להפעיל את החיוב, תקבלו את השגיאה הבאה: 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

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);
    }
}

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
}

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());
    }
  }
}

// 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`);
}

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']);
            }
        }
    }
}

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))

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

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

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());
    }
  }
}

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))

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

בשדות של תאריך ושעה, הפורמט של הנתונים בשיטה tabledata.insertAll הוא כדלקמן:

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

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

בדוגמה הבאה, השדה f_range_date מייצג עמודה RANGE<DATE> בטבלה. שורות מוכנסות לעמודה הזו באמצעות ה-API‏ tabledata.insertAll.

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

זמינות של נתונים בזמן אמת

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

לשורות שנוספו לאחרונה לטבלה עם חלוקה למחיצות לפי זמן ההוספה יש באופן זמני ערך 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 I/O של 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 עבור השאילתה הזו. לחלופין, אפשר לבחור עמודות ספציפיות כדי להשמיט את העמודה הזו.
• כדי לשלוח שאילתות לנתונים פעילים עם כפילויות שהוסרו, אפשר גם ליצור תצוגה מעל הטבלה באמצעות השאילתה להסרת כפילויות. חשוב לדעת שעלויות השאילתות בתצוגה מחושבות על סמך העמודות שנבחרו בתצוגה, ולכן יכול להיות שגודל הנתונים שנסרקו יהיה גדול.

הזרמה לטבלאות מחולקות למחיצות (Partitions) לפי זמן

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

חלוקה למחיצות (partitioning) לפי זמני כתיבת הנתונים

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

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

אם אתם מעבירים נתונים בסטרימינג לטבלה עם חלוקה למחיצות לפי יום, אתם יכולים לבטל את ההיסק של התאריך על ידי ציון מעצב מחיצות כחלק מהבקשה insertAll. צריך לכלול את המעצב בפרמטר tableId. לדוגמה, אתם יכולים להעביר בסטרימינג למחיצה שמתאימה לתאריך 2021-03-01 בטבלה table1 באמצעות מעצב המחיצות:

table1$20210301

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

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

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

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

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

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

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

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

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

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

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

<targeted_table_name> + <templateSuffix>

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

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

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

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

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

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

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

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

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

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

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

מכסה

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

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

משך החיים (TTL)

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

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

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

תצוגות

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

פתרון בעיות שקשורות להוספת מודעות לסטרימינג

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

קודי תגובת 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 ריק, יכול להיות שתופיע שגיאת המכסה הבאה:

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

המטרה של השדה 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% מהמכסה, סביר להניח שהשגיאות מצביעות על עליות זמניות. כדי לטפל בשגיאות האלה, צריך לנסות שוב את הפעולה באמצעות השהיה מעריכית לפני ניסיון חוזר בין הניסיונות החוזרים.

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

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

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