העלאות בסטרימינג

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

שימוש באימות סיכום ביקורת (checksum) במהלך סטרימינג

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

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

דרישות מוקדמות

הדרישות המוקדמות עשויות להשתנות בהתאם לכלי שבו נעשה שימוש:

המסוף

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

רשימת ההרשאות הנדרשות לפעולות ספציפיות מופיעה במאמר הרשאות IAM במסוף Google Cloud.

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

שורת הפקודה

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

רשימת ההרשאות הנדרשות לפעולות ספציפיות מופיעה במאמר הרשאות IAM לפקודות gsutil.

ספריות לקוח

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

אם לא צוין אחרת, בקשות לספריות לקוח מבוצעות דרך API בפורמט JSON ומחייבות את ההרשאות שמפורטות בהרשאות IAM ל-methods של JSON. כדי לראות אילו methods של API בפורמט JSON מופעלות כששולחים בקשות באמצעות ספריית לקוח, אתם צריכים לבצע רישום ביומן של הבקשות הגולמיות.

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

‫API בארכיטקטורת REST

‫API בפורמט JSON

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

רשימת ההרשאות הנדרשות לפעולות ספציפיות מופיעה במאמר הרשאות IAM ל-methods של JSON.

שידור של העלאה

הדוגמאות הבאות מראות איך לבצע העלאה בסטרימינג מתהליך לאובייקט של Cloud Storage:

המסוף

מסוף Google Cloud לא תומך בהעלאות בסטרימינג. במקום זאת, משתמשים ב-CLI של gcloud.

שורת הפקודה

gcloud

מעבירים את הנתונים לפקודה gcloud storage cp ומשתמשים במקף בשביל כתובת ה-URL של המקור:
```
PROCESS_NAME | gcloud storage cp - gs://BUCKET_NAME/OBJECT_NAME
```
כאשר:
- ‫PROCESS_NAME הוא שם התהליך שממנו אוספים נתונים. לדוגמה, collect_measurements.
- ‫BUCKET_NAME הוא שם הקטגוריה שמכילה את האובייקט. לדוגמה, my_app_bucket.
- ‫OBJECT_NAME הוא שם האובייקט שנוצר מהנתונים. לדוגמה, data_measurements.

gsutil

חשוב: gsutil הוא לא ה-CLI המומלץ לעבודה עם Cloud Storage. במקום זאת, משתמשים בפקודות gcloud storage ב-Google Cloud CLI.

מעבירים את הנתונים לפקודה gsutil cp ומשתמשים במקף בשביל כתובת ה-URL של המקור:
```
PROCESS_NAME | gsutil cp - gs://BUCKET_NAME/OBJECT_NAME
```
כאשר:
- ‫PROCESS_NAME הוא שם התהליך שממנו אוספים נתונים. לדוגמה, collect_measurements.
- ‫BUCKET_NAME הוא שם הקטגוריה שמכילה את האובייקט. לדוגמה, my_app_bucket.
- ‫OBJECT_NAME הוא שם האובייקט שנוצר מהנתונים. לדוגמה, data_measurements.

ספריות לקוח

C++

מידע נוסף מופיע במאמרי העזרה של Cloud Storage C++ API.

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

אל הקוד ב-GitHub משוב

namespace gcs = ::google::cloud::storage;
using ::google::cloud::StatusOr;
[](gcs::Client client, std::string const& bucket_name,
   std::string const& object_name, int desired_line_count) {
  std::string const text = "Lorem ipsum dolor sit amet";
  gcs::ObjectWriteStream stream =
      client.WriteObject(bucket_name, object_name);

  for (int lineno = 0; lineno != desired_line_count; ++lineno) {
    // Add 1 to the counter, because it is conventional to number lines
    // starting at 1.
    stream << (lineno + 1) << ": " << text << "\n";
  }

  stream.Close();

  StatusOr<gcs::ObjectMetadata> metadata = std::move(stream).metadata();
  if (!metadata) throw std::move(metadata).status();
  std::cout << "Successfully wrote to object " << metadata->name()
            << " its size is: " << metadata->size()
            << "\nFull metadata: " << *metadata << "\n";
}

C#

מידע נוסף מופיע במאמרי העזרה של Cloud Storage C# API.

אל הקוד ב-GitHub משוב


using Google.Cloud.Storage.V1;
using System;
using System.IO;

public class UploadFileSample
{
    public void UploadFile(
        string bucketName = "your-unique-bucket-name",
        string localPath = "my-local-path/my-file-name",
        string objectName = "my-file-name")
    {
        var storage = StorageClient.Create();
        using var fileStream = File.OpenRead(localPath);
        storage.UploadObject(bucketName, objectName, null, fileStream);
        Console.WriteLine($"Uploaded {objectName}.");
    }
}

Go

מידע נוסף מופיע במאמרי העזרה של Cloud Storage Go API.

אל הקוד ב-GitHub משוב

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"time"

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

// streamFileUpload uploads an object via a stream.
func streamFileUpload(w io.Writer, bucket, object string) error {
	// bucket := "bucket-name"
	// object := "object-name"
	ctx := context.Background()
	client, err := storage.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("storage.NewClient: %w", err)
	}
	defer client.Close()

	b := []byte("Hello world.")
	buf := bytes.NewBuffer(b)

	ctx, cancel := context.WithTimeout(ctx, time.Second*50)
	defer cancel()

	// Upload an object with storage.Writer.
	wc := client.Bucket(bucket).Object(object).NewWriter(ctx)
	wc.ChunkSize = 0 // note retries are not supported for chunk size 0.

	if _, err = io.Copy(wc, buf); err != nil {
		return fmt.Errorf("io.Copy: %w", err)
	}
	// Data can continue to be added to the file until the writer is closed.
	if err := wc.Close(); err != nil {
		return fmt.Errorf("Writer.Close: %w", err)
	}
	fmt.Fprintf(w, "%v uploaded to %v.\n", object, bucket)

	return nil
}

Java

מידע נוסף מופיע במאמרי העזרה של Cloud Storage Java API.

אל הקוד ב-GitHub משוב


import com.google.cloud.WriteChannel;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class StreamObjectUpload {

  public static void streamObjectUpload(
      String projectId, String bucketName, String objectName, String contents) throws IOException {
    // The ID of your GCP project
    // String projectId = "your-project-id";

    // The ID of your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    // The ID of your GCS object
    // String objectName = "your-object-name";

    // The string of contents you wish to upload
    // String contents = "Hello world!";

    Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();
    BlobId blobId = BlobId.of(bucketName, objectName);
    BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
    byte[] content = contents.getBytes(StandardCharsets.UTF_8);
    try (WriteChannel writer = storage.writer(blobInfo)) {
      writer.write(ByteBuffer.wrap(content));
      System.out.println(
          "Wrote to " + objectName + " in bucket " + bucketName + " using a WriteChannel.");
    }
  }
}

Node.js

מידע נוסף מופיע במאמרי העזרה של Cloud Storage Node.js API.

אל הקוד ב-GitHub משוב

/**
 * TODO(developer): Uncomment the following lines before running the sample
 */
// The ID of your GCS bucket
// const bucketName = 'your-unique-bucket-name';

// The new ID for your GCS file
// const destFileName = 'your-new-file-name';

// The content to be uploaded in the GCS file
// const contents = 'your file content';

// Imports the Google Cloud client library
const {Storage} = require('@google-cloud/storage');

// Import Node.js stream
const stream = require('stream');

// Creates a client
const storage = new Storage();

// Get a reference to the bucket
const myBucket = storage.bucket(bucketName);

// Create a reference to a file object
const file = myBucket.file(destFileName);

// Create a pass through stream from a string
const passthroughStream = new stream.PassThrough();
passthroughStream.write(contents);
passthroughStream.end();

async function streamFileUpload() {
  passthroughStream.pipe(file.createWriteStream()).on('finish', () => {
    // The file upload is complete
  });

  console.log(`${destFileName} uploaded to ${bucketName}`);
}

streamFileUpload().catch(console.error);

PHP

מידע נוסף מופיע במאמרי העזרה של Cloud Storage PHP API.

אל הקוד ב-GitHub משוב

use Google\Cloud\Storage\StorageClient;
use Google\Cloud\Storage\WriteStream;

/**
 * Upload a chunked file stream.
 *
 * @param string $bucketName The name of your Cloud Storage bucket.
 *        (e.g. 'my-bucket')
 * @param string $objectName The name of your Cloud Storage object.
 *        (e.g. 'my-object')
 * @param string $contents The contents to upload via stream chunks.
 *        (e.g. 'these are my contents')
 */
function upload_object_stream(string $bucketName, string $objectName, string $contents): void
{
    $storage = new StorageClient();
    $bucket = $storage->bucket($bucketName);
    $writeStream = new WriteStream(null, [
        'chunkSize' => 1024 * 256, // 256KB
    ]);
    $uploader = $bucket->getStreamableUploader($writeStream, [
        'name' => $objectName,
    ]);
    $writeStream->setUploader($uploader);
    $stream = fopen('data://text/plain,' . $contents, 'r');
    while (($line = stream_get_line($stream, 1024 * 256)) !== false) {
        $writeStream->write($line);
    }
    $writeStream->close();

    printf('Uploaded %s to gs://%s/%s' . PHP_EOL, $contents, $bucketName, $objectName);
}

Python

מידע נוסף מופיע במאמרי העזרה של Cloud Storage Python API.

אל הקוד ב-GitHub משוב

from google.cloud import storage

def upload_blob_from_stream(bucket_name, file_obj, destination_blob_name):
    """Uploads bytes from a stream or other file-like object to a blob."""
    # The ID of your GCS bucket
    # bucket_name = "your-bucket-name"

    # The stream or file (file-like object) from which to read
    # import io
    # file_obj = io.BytesIO()
    # file_obj.write(b"This is test data.")

    # The desired name of the uploaded GCS object (blob)
    # destination_blob_name = "storage-object-name"

    # Construct a client-side representation of the blob.
    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Rewind the stream to the beginning. This step can be omitted if the input
    # stream will always be at a correct position.
    file_obj.seek(0)

    # Upload data from the stream to your bucket.
    blob.upload_from_file(file_obj)

    print(
        f"Stream data uploaded to {destination_blob_name} in bucket {bucket_name}."
    )

Ruby

מידע נוסף מופיע במאמרי העזרה של Cloud Storage Ruby API.

אל הקוד ב-GitHub משוב


# The ID of your GCS bucket
# bucket_name = "your-unique-bucket-name"

# The stream or file (file-like object) from which to read
# local_file_obj = StringIO.new "This is test data."

# Name of a file in the Storage bucket
# file_name   = "some_file.txt"

require "google/cloud/storage"

storage = Google::Cloud::Storage.new
bucket  = storage.bucket bucket_name

local_file_obj.rewind
bucket.create_file local_file_obj, file_name

puts "Stream data uploaded to #{file_name} in bucket #{bucket_name}"

‫API בארכיטקטורת REST

‫API בפורמט JSON

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

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

לדוגמה, אם גודל המקטע הראשון שמעלים הוא 512KiB, הכותרת Content-Range של המקטע היא bytes 0-524287/*. אם בהעלאה נותרו 64,000 בייטים אחרי המקטע הראשון, צריך לשלוח מקטע סופי שמכיל את הבייטים הנותרים ויש לו כותרת Content-Range עם הערך bytes 524288-588287/588288.

‫API בפורמט XML

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

העלאה מרובת חלקים של API בפורמט XML.
העלאה שניתן להמשיך עם ההתאמות הבאות:
- כשמעלים את נתוני הקובץ עצמו, משתמשים בהעלאה של מספר מקטעים.
- מכיוון שאי אפשר לדעת מה גודל הקובץ הכולל עד שמגיעים למקטע הסופי, צריך להשתמש ב-* בשביל גודל הקובץ הכולל בכותרת Content-Range של מקטעי הביניים.
  
  לדוגמה, אם גודל המקטע הראשון שמעלים הוא 512KiB, הכותרת Content-Range של המקטע היא bytes 0-524287/*. אם בהעלאה נותרו 64,000 בייטים אחרי המקטע הראשון, צריך לשלוח מקטע סופי שמכיל את הבייטים הנותרים ויש לו כותרת Content-Range עם הערך bytes 524288-588287/588288.

העלאות בסטרימינג

שימוש באימות סיכום ביקורת (checksum) במהלך סטרימינג

דרישות מוקדמות

המסוף

שורת הפקודה

ספריות לקוח

‫API בארכיטקטורת REST

‫API בפורמט JSON

שידור של העלאה

המסוף

שורת הפקודה

gcloud

gsutil

ספריות לקוח

C++

C#

Go

Java

Node.js

PHP

Python

Ruby

‫API בארכיטקטורת REST

‫API בפורמט JSON

‫API בפורמט XML

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