‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

ניהול שידורים חיים

באמצעות Datastream API אתם יכולים להפוך את תהליכי העבודה של שכפול הנתונים לאוטומטיים, באמצעות קריאות ל-API בארכיטקטורת REST או Google Cloud CLI. אתם יכולים ליצור, לאמת, לשנות ולמחוק מקורות נתונים, או לאחזר ולהציג מקורות נתונים ואובייקטים של מקורות נתונים.

אתם יכולים גם להתחיל, להשהות, להמשיך או לשחזר זרמי נתונים, לנהל מילוי חוזר של אובייקטים, לשנות את מספר המשימות שפועלות בו-זמנית ולהעביר אובייקטים גדולים של Oracle. לפרטים על הפקודות, ראו gcloud datastream streams. במאמר שימוש בממשקי Datastream API מוסבר איך עושים את זה.

יצירת מקור נתונים

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

הקטע הזה עוסק בתרחישים הבאים לדוגמה:

העברה בסטרימינג מ-Oracle ל-Cloud Storage
העברה בסטרימינג מ-MySQL ל-BigQuery
סטרימינג מ-PostgreSQL ל-BigQuery
הגדרת קבוצת אובייקטים שייכללו בסטרימינג
מילוי חוסרים בכל האובייקטים שכלולים בסטרימינג
החרגת אובייקטים מהסטרימינג
החרגת אובייקטים מהמילוי החוזר
הגדרת CMEK להצפנת נתונים במנוחה
הגדרת מצב כתיבה לסטרימינג
העברת נתונים לפרויקט אחר ב-BigQuery
העברת נתונים בסטרימינג לטבלאות Apache Iceberg
העברה בסטרימינג מ-MongoDB באמצעות מצב JSON מחמיר

דוגמה 1: העברת אובייקטים ספציפיים בסטרימינג ל-BigQuery

בדוגמה הזו נסביר איך:

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

הבקשה הבאה היא לשליפת כל הטבלאות מ-schema1 ושתי טבלאות ספציפיות מ-schema2: tableA ו-tableC. האירועים נכתבים במערך נתונים ב-BigQuery.

הבקשה לא כוללת את הפרמטר customerManagedEncryptionKey, ולכן המערכת הפנימית של Google Cloud לניהול מפתחות משמשת להצפנת הנתונים במקום CMEK.

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

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlCdcStream
{
  "displayName": "MySQL CDC to BigQuery",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "schema1" },
          {
            "database": "schema2",
            "mysqlTables": [
              {
                "table": "tableA",
                "table": "tableC"
              }
            ]
          }
        ]
      },
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "dataFreshness": "900s"
    }
  },
  "backfillAll": {}
}

gcloud

למידע נוסף על השימוש ב-gcloud ליצירת סטרימינג, קראו את המסמכים של Google Cloud SDK.

דוגמה 2: החרגת אובייקטים ספציפיים מ-stream עם מקור PostgreSQL

בדוגמה הזו נסביר איך:

העברת נתונים בסטרימינג מ-PostgreSQL ל-BigQuery
החרגת אובייקטים מהסטרימינג
החרגת אובייקטים מהמילוי החוזר

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

‫replicationSlot: משבצת רפליקציה היא דרישה מוקדמת להגדרת מסד נתונים של PostgreSQL לרפליקציה. צריך ליצור משבצת רפליקציה לכל מקור נתונים.
‫publication: פרסום הוא קבוצת טבלאות שרוצים לשכפל את השינויים שלהן. שם אתר החדשות חייב להיות קיים במסד הנתונים לפני שמתחילים שידור. הפרסום חייב לכלול לפחות את הטבלאות שצוינו ברשימת includeObjects של הסטרימינג.

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

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myPostgresStream
{
  "displayName": "PostgreSQL to BigQueryCloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/postgresCp",
    "postgresqlSourceConfig": {
      "replicationSlot": "replicationSlot1",
      "publication": "publicationA",
      "includeObjects": {
        "postgresqlSchemas": {
          "schema": "schema1"
        }
      },
      "excludeObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
        "postgresqlTables": [
          {
            "table": "tableA",
            "postgresqlColumns": [
              { "column": "column5" }
              ]
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "dataFreshness": "900s",
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
           "location": "us",
           "datasetIdPrefix": "prefix_"
        }
      }
    }
  },
  "backfillAll": {
    "postgresqlExcludedObjects": {
        "postgresqlSchemas": [
          { "schema": "schema1",
            "postgresqlTables": [
              { "table": "tableA" }
            ]
          }
        ]
      }
    }
  }

gcloud

למידע נוסף על השימוש ב-gcloud ליצירת סטרימינג, קראו את המסמכים של Google Cloud SDK.

דוגמה 3: ציון מצב כתיבה של הוספה בלבד לסטרימינג

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

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

הבקשה הבאה מראה איך להגדיר את המצב appendOnly כשיוצרים זרם מ-MySQL ל-BigQuery.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=appendOnlyStream
{
  "displayName": "My append-only stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        }
      },
      "appendOnly": {}
    }
  },
  "backfillAll": {}
}

gcloud

למידע נוסף על השימוש ב-gcloud ליצירת סטרימינג, קראו את המסמכים של Google Cloud SDK.

דוגמה 4: סטרימינג לפרויקט אחר ב-BigQuery

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

אם מציינים את הערך sourceHierarchyDatasets במערך הנתונים של היעד, צריך לאכלס את השדה projectId.

אם מציינים את הערך singleTargetDataset עבור קבוצת הנתונים של היעד, צריך לאכלס את השדה datasetId בפורמט projectId:datasetId.

REST

עבור sourceHierarchyDatasets:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream1
{
  "displayName": "My cross-project stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "sourceHierarchyDatasets": {
        "datasetTemplate": {
          "location": "us",
          "datasetIdPrefix": "prefix_"
        },
        "projectId": "myProjectId2"
      }
    }
  },
  "backfillAll": {}
}

עבור singleTargetDataset:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=crossProjectBqStream2
{
  "displayName": "My cross-project stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          { "database": "myMySqlDb"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "BigQueryCp",
    "bigqueryDestinationConfig": {
      "singleTargetDataset": {
        "datasetId": "myProjectId2:myDatasetId"
      },
    }
  },
  "backfillAll": {}
}

gcloud

עבור sourceHierarchyDatasets:

  datastream streams create crossProjectBqStream1 --location=us-central1
  --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json
  --destination=destination-cp --bigquery-destination-config=source_hierarchy_cross_project_config.json
  --backfill-none

התוכן של קובץ התצורה source_hierarchy_cross_project_config.json:

  {"sourceHierarchyDatasets": {"datasetTemplate": {"location": "us-central1", "datasetIdPrefix": "prefix_"}, "projectId": "myProjectId2"}}

עבור singleTargetDataset:

  datastream streams create crossProjectBqStream --location=us-central1
  --display-name=my-cross-project-stream --source=source-cp --mysql-source-config=mysql_source_config.json
  --destination=destination-cp --bigquery-destination-config=single_target_cross_project_config.json
  --backfill-none

התוכן של קובץ התצורה single_target_cross_project_config.json:

  {"singleTargetDataset": {"datasetId": "myProjectId2:myDatastetId"}}

למידע נוסף על השימוש ב-gcloud ליצירת סטרימינג, קראו את המסמכים של Google Cloud SDK.

דוגמה 5: סטרימינג ליעד ב-Cloud Storage

בדוגמה הזו נסביר איך:

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

הבקשה הבאה מראה איך ליצור זרם שכותב את האירועים לקטגוריה ב-Cloud Storage.

בבקשה לדוגמה הזו, האירועים נכתבים בפורמט הפלט JSON, ונוצר קובץ חדש בכל 100MB או 30 שניות (כך שערכי ברירת המחדל של 50MB ו-60 שניות מוחלפים).

בפורמט JSON, אתם יכולים:

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

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

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

מידע נוסף על אוספי מפתחות מופיע במאמר משאבי Cloud KMS. מידע נוסף על הגנה על הנתונים באמצעות מפתחות הצפנה זמין במאמר Cloud Key Management Service (KMS)‎.

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleCdcStream
{
  "displayName": "Oracle CDC to Cloud Storage",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/
    connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "GcsBucketCp",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "jsonFileFormat": {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      },
      "fileRotationMb": 100,
      "fileRotationInterval": 30
    }
  },
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillNone": {}
}

gcloud

למידע נוסף על השימוש ב-gcloud ליצירת סטרימינג, קראו את המסמכים של Google Cloud SDK.

דוגמה 6: העברה בסטרימינג לטבלת Apache Iceberg

בדוגמה הזו נסביר איך להגדיר העברה כדי לשכפל נתונים ממסד נתונים של MySQL לטבלת Apache Iceberg במצב append-only. לפני שיוצרים את הבקשה, צריך לבצע את הפעולות הבאות:

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

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

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams?streamId=mysqlIcebergStream
{
  "displayName": "MySQL to Apache Iceberg stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/streams/mysqlIcebergCp",
    "mysqlSourceConfig": {
      "includeObjects": {
        "mysqlDatabases": [
          {
            "database": "my-mysql-database"
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id",
    "bigqueryDestinationConfig": {
      "blmtConfig": {
        "bucket": "my-gcs-bucket-name",
        "rootPath": "my/folder",
        "connectionName": "my-project-id.us-central1.my-bigquery-connection-name",
        "fileFormat": "PARQUET",
        "tableFormat": "ICEBERG"
        },
      "singleTargetDataset": {
        "datasetId": "my-project-id:my-bigquery-dataset-id"
      },
      "appendOnly": {}
    }
  },
  "backfillAll": {}
}

gcloud

datastream streams create mysqlIcebergStream --location=us-central1
--display-name=mysql-to-bl-stream --source=source --mysql-source-config=mysql_source_config.json
--destination=destination --bigquery-destination-config=bl_config.json
--backfill-none

התוכן של קובץ התצורה של המקור mysql_source_config.json:

{"excludeObjects": {}, "includeObjects": {"mysqlDatabases":[{"database":"my-mysql-database"}]}}

התוכן של קובץ התצורה bl_config.json:

{ "blmtConfig": { "bucket": "my-gcs-bucket-name", "rootPath": "my/folder", "connectionName": "my-project-id.us-central1.my-bigquery-connection-name", "fileFormat": "PARQUET", "tableFormat": "ICEBERG" }, "singleTargetDataset": {"datasetId": "my-project-id:my-bigquery-dataset-id"}, "appendOnly": {} }

Terraform

resource "google_datastream_stream" "stream" {
  stream_id    = "mysqlBlStream"
  location     = "us-central1"
  display_name = "MySQL to Apache Iceberg stream"

  source_config {
    source_connection_profile = "/projects/myProjectId1/locations/us-central1/streams/mysqlBlCp"
    mysql_source_config {
      include_objects {
        mysql_databases {
          database = "my-mysql-database"
        }
      }
    }
  }

  destination_config {
    destination_connection_profile = "projects/myProjectId1/locations/us-central1/connectionProfiles/my-bq-cp-id"
    bigquery_destination_config {
      single_target_dataset {
        dataset_id = "my-project-id:my-bigquery-dataset-id"
      }
      blmt_config {
        bucket          = "my-gcs-bucket-name"
        table_format    = "ICEBERG"
        file_format     = "PARQUET"
        connection_name = "my-project-id.us-central1.my-bigquery-connection-name"
        root_path       = "my/folder"
      }
      append_only {}
    }
  }

  backfill_none {}
}

דוגמה 7: סטרימינג מ-MongoDB באמצעות מצב JSON מחמיר

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

כברירת מחדל, Datastream משתמש במצב JSON קנוני (CANONICAL) כדי לתת עדיפות לנאמנות הנתונים. אם האפליקציות במורד הזרם צריכות להיות תואמות למנתחי JSON רגילים, אתם יכולים לציין מצב JSON מחמיר (STRICT).

כשיוצרים מקור נתונים ממקור MongoDB:

הערכים התקינים של jsonMode ב-API הם CANONICAL (ברירת מחדל) ו-STRICT.
ב-CLI של gcloud אין דגל נפרד למצב JSON. במקום זאת, הגדר את הגדרות מקור MongoDB באמצעות קובץ תצורה שמועבר לדגל --mongodb-source-config.

REST

ב-API בארכיטקטורת REST של Datastream, מגדירים את jsonMode לערך "STRICT" באובייקט sourceConfig.mongodbSourceConfig:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID
{
  "displayName": "STREAM_DISPLAY_NAME",
  "sourceConfig": {
    "sourceConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE",
    "mongodbSourceConfig": {
      "jsonMode": "STRICT",
      "includeObjects": {
        "databases": [
          {
            "database": "DATABASE_NAME",
            "collections": [
              {
                "collection": "COLLECTION_NAME"
              }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfile": "projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE",
    "bigqueryDestinationConfig": {
      "singleTargetDataset": {
        "datasetId": "PROJECT_ID:DATASET_ID"
      }
    }
  }
}

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

‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
‫LOCATION: האזור שבו נוצר ה-stream, למשל us-central1.
‫STREAM_ID: המזהה של הסטרימינג.
‫STREAM_DISPLAY_NAME: שם ידידותי למשתמש של הסטרימינג.
‫SOURCE_PROFILE: המזהה של פרופיל החיבור של MongoDB במקור.
‫DEST_PROFILE: המזהה של פרופיל החיבור ליעד.
‫DATABASE_NAME: השם של מסד הנתונים באשכול המקור ב-MongoDB.
‫COLLECTION_NAME: שם האוסף במסד הנתונים.
‫DATASET_ID: המזהה של מערך הנתונים של היעד ב-BigQuery.

gcloud

כדי ליצור סטרימינג עם מקור MongoDB ומצב JSON מחמיר באמצעות ה-CLI של gcloud:

יוצרים קובץ תצורה של מקור בפורמט JSON או YAML, ומציינים את "jsonMode": "STRICT" ואת האובייקטים שרוצים לשכפל.

קובץ תצורה בפורמט JSON, לדוגמה, mongo_source_config.json:

{
  "jsonMode": "STRICT",
  "includeObjects": {
    "databases": [
      {
        "database": "DATABASE_NAME",
        "collections": [
          {
            "collection": "COLLECTION_NAME"
          }
        ]
      }
    ]
  }
}

קובץ תצורה של YAML, לדוגמה, mongo_source_config.yaml:

jsonMode: STRICT
includeObjects:
  databases:
    - database: DATABASE_NAME
      collections:
        - collection: COLLECTION_NAME

מעבירים את קובץ התצורה לדגל --mongodb-source-config כשיוצרים את הסטרימינג:

gcloud datastream streams create STREAM_ID \
    --location=LOCATION \
    --display-name="STREAM_DISPLAY_NAME" \
    --source=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/SOURCE_PROFILE \
    --destination=projects/PROJECT_ID/locations/LOCATION/connectionProfiles/DEST_PROFILE \
    --mongodb-source-config=mongo_source_config.json \
    --bigquery-destination-config=bigquery_destination_config.json

התוכן של קובץ התצורה bigquery_destination_config.json:

{
  "singleTargetDataset": {
    "datasetId": "PROJECT_ID:DATASET_ID"
  }
}

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

‫STREAM_ID: המזהה של הסטרימינג.
‫LOCATION: האזור שבו נוצר ה-stream, למשל us-central1.
‫STREAM_DISPLAY_NAME: שם ידידותי למשתמש של הסטרימינג.
‫PROJECT_ID: מזהה הפרויקט ב- Google Cloud .
‫SOURCE_PROFILE: המזהה של פרופיל החיבור ל-MongoDB במקור.
‫DEST_PROFILE: המזהה של פרופיל החיבור ליעד.
‫DATABASE_NAME: השם של מסד הנתונים באשכול המקור ב-MongoDB.
‫COLLECTION_NAME: שם האוסף במסד הנתונים.
‫DATASET_ID: המזהה של מערך הנתונים של היעד ב-BigQuery.

למידע נוסף על השימוש ב-gcloud ליצירת מקור נתונים, עיינו במסמכי התיעוד של Google Cloud SDK.

אימות ההגדרה של מקור נתונים

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

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

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

כדי לאמת שידור, מוסיפים את &validate_only=true לכתובת ה-URL שלפני גוף הבקשה:

POST "https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams?streamId=STREAM_ID&validate_only=true"

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

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

CMEK_DOES_NOT_EXIST_OR_MISSING_PERMISSIONS

כדי לפתור את הבעיה, צריך לוודא שהמפתח שסיפקתם קיים, ושלחשבון השירות של Datastream יש את ההרשאה cloudkms.cryptoKeys.get למפתח.

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

קבלת מידע על שידור

הקוד הבא מציג בקשה לאחזור מידע על שידור. המידע כולל:

שם מקור הנתונים (מזהה ייחודי)
שם ידידותי למשתמש של השידור (שם לתצוגה)
חותמות זמן של מועד היצירה של מקור הנתונים ומועד העדכון האחרון שלו
מידע על פרופילי החיבור של המקור והיעד שמשויכים ל-stream
מצב השידור

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/streams/STREAM_ID

התגובה שמופיעה נראית כך:

{
  "name": "myOracleCdcStream",
  "displayName": "Oracle CDC to Cloud Storage",
  "createTime": "2019-12-15T15:01:23.045123456Z",
  "updateTime": "2019-12-15T15:01:23.045123456Z",
  "sourceConfig": {
    "sourceConnectionProfileName": "myOracleDb",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1"
          },
          {
            "schema": "schema3",
            "oracleTables": [
              { "table": "tableA" },
              { "table": "tableC" }
            ]
          }
        ]
      }
    }
  },
  "destinationConfig": {
    "destinationConnectionProfileName": "myGcsBucket",
    "gcsDestinationConfig": {
      "path": "/folder1",
      "avroFileFormat": {},
      "fileRotationMb": 100,
      "fileRotationInterval": 60
    }
  },
  "state": "RUNNING"
  "customerManagedEncryptionKey": "projects/myProjectId1/locations/us-central1/
  keyRings/myRing/cryptoKeys/myEncryptionKey",
  "backfillAll": {}
}

gcloud

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

הצגת רשימת מקורות הנתונים

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

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams

gcloud

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

הצגת רשימה של אובייקטים בסטרימינג

הקוד הבא מציג בקשה לאחזור מידע על כל האובייקטים של שידור.

REST

GET https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects

gcloud

מידע נוסף על השימוש ב-gcloud כדי לאחזר מידע על כל האובייקטים של הסטרימינג זמין במסמכי התיעוד של Google Cloud SDK.

רשימת האובייקטים שמוחזרת עשויה להיראות כך:

REST

{
  "streamObjects": [
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object1",
      "displayName": "employees.salaries",
      "backfillJob": {
        "state": "ACTIVE",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T12:12:26.344878Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "salaries"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object2",
      "displayName": "contractors.hours",
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "contractors",
          "table": "hours"
        }
      }
    },
    {
      "name": "projects/myProjectId1/locations/us-central1/streams/myStream/
      objects/object3",
      "displayName": "employees.departments",
      "backfillJob": {
        "state": "COMPLETED",
        "trigger": "AUTOMATIC",
        "lastStartTime": "2021-10-18T11:26:12.869880Z",
        "lastEndTime": "2021-10-18T11:26:28.405653Z"
      },
      "sourceObject": {
        "mysqlIdentifier": {
          "database": "employees",
          "table": "departments"
        }
      }
    }
  ]
}

gcloud

מידע נוסף על השימוש ב-gcloud כדי להציג רשימה של אובייקטים בסטרימינג זמין במסמכי התיעוד של Google Cloud SDK.

התחלת שידור

הקוד הבא מציג בקשה להתחלת שידור.

כשמשתמשים בפרמטר updateMask בבקשה, צריך לכלול בגוף הבקשה רק את השדות שציינתם. כדי להתחיל שידור, משנים את הערך בשדה state מ-CREATED ל-RUNNING.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

מידע נוסף על השימוש ב-gcloud כדי להתחיל את הסטרימינג זמין במסמכי התיעוד של Google Cloud SDK.

השהיית שידור

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

בדוגמה הזו, השדה שצוין בפרמטר updateMask הוא השדה state. השהיית השידור משנה את המצב שלו מ-RUNNING ל-PAUSED.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "PAUSED"
}

gcloud

מידע נוסף על השימוש ב-gcloud כדי להשהות את הסטרימינג זמין במסמכי התיעוד של Google Cloud SDK.

המשך השידור

הקוד הבא מציג בקשה להמשך שידור שהושהה.

בדוגמה הזו, השדה שצוין בפרמטר updateMask הוא השדה state. המשך השידור ישנה את הסטטוס שלו מ-PAUSED ל-RUNNING.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=state
{
  "state": "RUNNING"
}

gcloud

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

שחזור שידור

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

שחזור של מקור נתונים ב-MySQL או ב-Oracle

בדוגמאות הקוד הבאות מוצגות בקשות לשחזור של זרם נתונים ממקור MySQL או Oracle ממיקומים שונים בקובץ היומן:

REST

לשחזר שידור מהמיקום הנוכחי. זוהי אפשרות ברירת המחדל:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

כדי לשחזר שידור מהמיקום הזמין הבא:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "nextAvailableStartPosition": {}
  }
}

כדי לשחזר שידור מהמיקום האחרון:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "mostRecentStartPosition": {}
  }
}

שחזור של סטרימינג ממיקום ספציפי (רפליקציה מבוססת-binlog של MySQL):

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

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

‫NAME_OF_THE_LOG_FILE: השם של קובץ היומן שממנו רוצים לשחזר את הסטרימינג
‫POSITION: המיקום בקובץ היומן שממנו רוצים לשחזר את הסטרימינג. אם לא תספקו את הערך, Datastream ישחזר את הסטרימינג מתחילת הקובץ.

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 4
      }
    }
  }
}

שחזור של סטרימינג ממיקום ספציפי (רפליקציה מבוססת-GTID של MySQL):

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "GTID_SET"
      }
    }
  }
}

מחליפים את GTID_SET במזהה GTID אחד או יותר, או בטווחים של מזהי GTID, שמהם רוצים לשחזר את הסטרימינג.

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:1-3"
      }
    }
  }
}

שחזור שידור ממיקום ספציפי (Oracle):

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}

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

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 234234
      }
    }
  }
}

למידע נוסף על אפשרויות השחזור הזמינות, ראו שחזור של זרם נתונים.

gcloud

אי אפשר לשחזר שידור באמצעות gcloud.

שחזור של מקור נתונים ב-PostgreSQL

בדוגמת הקוד הבאה מוצגת בקשה לשחזור של זרם נתונים ממקור PostgreSQL. במהלך השחזור, ה-stream מתחיל לקרוא מהמספר הסידורי הראשון של היומן (LSN) במשבצת הרפליקציה שהוגדרה ל-stream.

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "nextAvailableStartPosition": {}
  }
}

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run

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

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.postgresqlSourceConfig.replicationSlot
{
  "sourceConfig": {
    "postgresqlSourceConfig": {
      "replicationSlot": "NEW_REPLICATION_SLOT_NAME"
    }
  }
}

gcloud

אי אפשר לשחזר שידור באמצעות gcloud.

שחזור של מקור SQL Server ב-Stream

בדוגמאות הקוד הבאות מוצגות בקשות לדוגמה לשחזור של זרם נתונים ממקור SQL Server.

REST

כדי לשחזר שידור מהמיקום הזמין הראשון:

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run

שחזור שידור ממספר סידורי מועדף ביומן:

POST https://datastream.googleapis.com/v1/projects/[project-id]/locations/
[location]/streams/[stream-id]:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": lsn
      }
    }
  }
}

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/mySqlServerStreamId:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": 0000123C:0000BA78:0004
      }
    }
  }
}

gcloud

אי אפשר לשחזר שידור באמצעות gcloud.

התחלה או המשך של שידור ממיקום ספציפי

אפשר להתחיל שידור או להמשיך שידור שהושהה ממיקום ספציפי במקורות MySQL,‏ Oracle ו-SQL Server. האפשרות הזו שימושית כשרוצים לבצע מילוי חוסרים באמצעות כלי חיצוני, או להתחיל CDC ממיקום שאתם מציינים:

במקור MySQL, צריך לציין מיקום של binlog או קבוצת GTID.
במקור של Oracle, צריך לציין מספר שינוי מערכת (SCN) בקובץ יומן ה-Redo.
במקור של SQL Server, צריך לציין מספר סידורי של יומן (LSN) בטבלת שינויים.

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

התחלה או המשך של שידור ממיקום ספציפי ב-binlog‏ (MySQL):

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "NAME_OF_THE_LOG_FILE"
        "logPosition": POSITION
      }
    }
  }
}

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

‫NAME_OF_THE_LOG_FILE: השם של קובץ היומן שממנו רוצים להתחיל את הסטרימינג.
‫POSITION: המיקום בקובץ היומן שממנו רוצים להתחיל את הסטרימינג. אם לא תספקו את הערך, Datastream יתחיל לקרוא מתחילת הקובץ.

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlLogPosition": {
        "logFile": "binlog.001"
        "logPosition": 2
      }
    }
  }
}

gcloud

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

התחלה או המשך של סטרימינג מקבוצת GTID ספציפית (MySQL):

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "GTID_SET"
      }
    }
  }
}

מחליפים את GTID_SET ב-GTID אחד או יותר, או בטווחים של GTID, שמהם רוצים להתחיל או להמשיך את הסטרימינג.

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "mysqlGtidPosition": {
        "gtidSet": "22cc56f5-3862-379a-9af5-051c59baef9d:1-561143685:561143688-591036613,b8d7df02-832b-32b9-bec7-2018806b76f6:3-7"
      }
    }
  }
}

gcloud

התחלה או המשך של סטרימינג ממספר שינוי מערכת ספציפי בקובץ יומן ה-Redo ‏ (Oracle):

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": scn
      }
    }
  }
}

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

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "oracleScnPosition": {
        "scn": 123123
      }
    }
  }
}

gcloud

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

התחלה או המשך של סטרימינג ממספר סידורי ספציפי ביומן (SQL Server):

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": "lsn"
      }
    }
  }
}

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

לדוגמה:

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams/myStreamId1:run
{
  "cdcStrategy": {
    "specificStartPosition": {
      "sqlServerLsnPosition": {
        "lsn": "0000123C0000BA780004"
      }
    }
  }
}

למידע נוסף, קראו את מאמרי העזרה של Datastream API.

gcloud

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

שינוי מקור נתונים

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

בדוגמה הזו, השדות שצוינו בפרמטר updateMask כוללים את השדות fileRotationMb ו-fileRotationInterval, שמיוצגים על ידי הדגלים destinationConfig.gcsDestinationConfig.fileRotationMb ו-destinationConfig.gcsDestinationConfig.fileRotationInterval, בהתאמה.

REST

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
fileRotationMb,destinationConfig.gcsDestinationConfig.fileRotationInterval
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "fileRotationMb": 75,
      "fileRotationInterval": 45
    }
  }
}

הקוד הבא מציג בקשה לכלול קובץ סכימה של Unified Types בנתיב של הקבצים ש-Datastream כותב ב-Cloud Storage. כתוצאה מכך, Datastream כותב שני קבצים: קובץ נתונים בפורמט JSON וקובץ סכימה בפורמט Avro.

בדוגמה הזו, השדה שצוין הוא השדה jsonFileFormat, שמיוצג על ידי הדגל destinationConfig.gcsDestinationConfig.jsonFileFormat.

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=destinationConfig.gcsDestinationConfig.
jsonFileFormat
{
  "destinationConfig": {
    "gcsDestinationConfig": {
      "jsonFileFormat" {
        "schemaFileFormat": "AVRO_SCHEMA_FILE"
      }  
    }
  }
}

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

בקטע oracleExcludedObjects בקוד מוצגות הטבלאות והסכימות שמוגבלות למילוי חוסרים ביעד.

בדוגמה הזו, כל הטבלאות והסכימות ימולאו במילוי חסר, חוץ מ-tableA ב-schema3.

PATCH https://datastream.googleapis.com/v1/projects/myProjectId1/locations/us-central1/streams/myStream?updateMask=backfillAll
{
  "backfillAll": {
    "oracleExcludedObjects": {
      "oracleSchemas": [
        {
          "schema": "schema3",
          "oracleTables": [
            { "table": "tableA" }
          ]
        }
      ]
    }
  }
}

gcloud

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

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

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

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

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

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

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

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

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

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:startBackfillJob

gcloud

מידע נוסף על השימוש ב-gcloud כדי להתחיל מילוי חוסרים של אובייקט בסטרימינג זמין במסמכי התיעוד של Google Cloud SDK.

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

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

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

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

REST

POST https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID/objects/OBJECT_ID:stopBackfillJob

gcloud

מידע נוסף על השימוש ב-gcloud כדי להפסיק את המילוי החסר של אובייקט בסטרימינג זמין במסמכי התיעוד של Google Cloud SDK.

שינוי המספר המקסימלי של משימות CDC בו-זמניות

הקוד הבא מראה איך להגדיר את המספר המקסימלי של משימות סימון נתונים שהשתנו (CDC) בו-זמנית ל-7 ב-MySQL stream.

בדוגמה הזו, השדה שצוין בפרמטר updateMask הוא השדה maxConcurrentCdcTasks. אם מגדירים את הערך ל-7, משנים את המספר המקסימלי של משימות CDC בו-זמניות מהערך הקודם ל-7. אפשר להשתמש בערכים מ-0 עד 50 (כולל). אם לא תגדירו את הערך, או אם תגדירו אותו כ-0, ברירת המחדל של המערכת תהיה 5 משימות בסטרים.

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentCdcTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentCdcTasks": "7"
      }
    }  
}

gcloud

מידע נוסף על השימוש ב-gcloud זמין במסמכי התיעוד של Google Cloud SDK.

שינוי המספר המקסימלי של משימות מילוי חוסרים בו-זמניות

הקוד הבא מראה איך להגדיר את המספר המקסימלי של משימות מילוי חוסרים בו-זמניות ל-25 בסטרימינג של MySQL.

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

REST

PATCH https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/
streams/STREAM_ID?updateMask=sourceConfig.mysqlSourceConfig.maxConcurrentBackfillTasks
{
  "sourceConfig": {
    "mysqlSourceConfig": {
      "maxConcurrentBackfillTasks": "25"
      }
    }  
}

gcloud

מידע נוסף על השימוש ב-gcloud זמין במסמכי התיעוד של Google Cloud SDK.

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

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

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

REST

POST https://datastream.googleapis.com/v1/projects/myProjectId1/locations/
us-central1/streams?streamId=myOracleLobStream
{
  "displayName": "Oracle LOB stream",
  "sourceConfig": {
    "sourceConnectionProfileName": "/projects/myProjectId1/locations/us-central1/connectionProfiles/OracleCp",
    "oracleSourceConfig": {
      "includeObjects": {
        "oracleSchemas": [
          {
            "schema": "schema1",
            "oracleTables": [
              {
                "table": "tableA",
                "oracleColumns": [
                  {
                    "column": "column1,column2"
                  }
                ]
              }
            ]
          }
        ]
      },
      "excludeObjects": {},
      "streamLargeObjects": {}
    }
  }
}

gcloud

מידע נוסף על השימוש ב-gcloud לעדכון של מקור נתונים זמין במסמכי התיעוד של Google Cloud SDK.

מחיקת מקור נתונים

הקוד הבא מציג בקשה למחיקת שידור.

REST

DELETE https://datastream.googleapis.com/v1/projects/PROJECT_ID/locations/
LOCATION/streams/STREAM_ID

gcloud

מידע נוסף על השימוש ב-gcloud למחיקת הסטרימינג זמין במסמכי התיעוד של Google Cloud SDK.

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

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

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

יצירת מקור נתונים

דוגמה 1: העברת אובייקטים ספציפיים בסטרימינג ל-BigQuery

REST

gcloud

דוגמה 2: החרגת אובייקטים ספציפיים מ-stream עם מקור PostgreSQL

REST

gcloud

דוגמה 3: ציון מצב כתיבה של הוספה בלבד לסטרימינג

REST

gcloud

דוגמה 4: סטרימינג לפרויקט אחר ב-BigQuery

REST

gcloud

דוגמה 5: סטרימינג ליעד ב-Cloud Storage

REST

gcloud

דוגמה 6: העברה בסטרימינג לטבלת Apache Iceberg

REST

gcloud

Terraform

דוגמה 7: סטרימינג מ-MongoDB באמצעות מצב JSON מחמיר

REST

gcloud

אימות ההגדרה של מקור נתונים

קבלת מידע על שידור

REST

gcloud

הצגת רשימת מקורות הנתונים

REST

gcloud

הצגת רשימה של אובייקטים בסטרימינג

REST

gcloud

REST

gcloud

התחלת שידור

REST

gcloud

השהיית שידור

REST

gcloud

המשך השידור

REST

gcloud

שחזור שידור

שחזור של מקור נתונים ב-MySQL או ב-Oracle

REST

gcloud

שחזור של מקור נתונים ב-PostgreSQL

REST

gcloud

שחזור של מקור SQL Server ב-Stream

REST

gcloud

התחלה או המשך של שידור ממיקום ספציפי

REST

gcloud

REST

gcloud

REST

gcloud

REST

gcloud

שינוי מקור נתונים

REST

gcloud

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

REST

gcloud

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

REST

gcloud

שינוי המספר המקסימלי של משימות CDC בו-זמניות

REST

gcloud

שינוי המספר המקסימלי של משימות מילוי חוסרים בו-זמניות

REST

gcloud

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

ניהול שידורים חיים