Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

יצירת חיבורים של סנכרון שינויים בזרמי נתונים אל Kafka

בדף הזה מוסבר איך להשתמש במחבר Kafka כדי לצרוך נתונים מסנכרון שינויים בזרמי נתונים של Spanner ולהעביר אותם.

מושגי ליבה

בהמשך מוסברים מושגי הליבה של מחבר Kafka.

Debezium

‫Debezium הוא פרויקט קוד פתוח שמספק פלטפורמה להזרמת נתונים עם השהיה נמוכה לסימון נתונים שהשתנו (CDC).

מחבר Kafka

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

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

פלט של מחבר Kafka

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

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

הקלטת הזמנה

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

שינוי אירועים

מחבר Kafka יוצר אירוע של שינוי נתונים לכל פעולה של INSERT, UPDATE ו-DELETE. כל אירוע מכיל מפתח וערכים לשורה שהשתנתה.

אפשר להשתמש בממירים של Kafka Connect כדי ליצור אירועים של שינוי נתונים בפורמטים Protobuf,‏ AVRO,‏ JSON או JSON Schemaless. אם משתמשים בממיר Kafka Connect שמפיק סכימות, האירוע מכיל סכימות נפרדות למפתח ולערכים. אחרת, האירוע יכיל רק את המפתח והערכים.

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

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

השדה הראשון schema מציין סכימת Kafka Connect שמתארת את סכימת המפתח של Spanner.
השדה הראשון payload כולל את המבנה שמתואר בשדה schema הקודם ומכיל את המפתח של השורה ששוּנתה.
השדה השני schema מציין את סכימת Kafka Connect שמתארת את הסכימה של השורה ששוּנתה.
השדה השני payload בנוי לפי המבנה שמתואר בשדה הקודם schema והוא מכיל את הנתונים בפועל של השורה שהשתנתה.
השדה source הוא שדה חובה שמתאר את מטא-נתוני המקור של האירוע.

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

{
  // The schema for the Spanner key.
  "schema": {
    "type": "struct",
    "name": "customers.Key",
    "optional": false,
    "fields": [
      {
        "type": "int64",
        "optional": "false"
        "field": "false"
      }
    ]
  },
  // The value of the Spanner key.
  "payload": {
      "id": "1"
  },
  // The schema for the payload, which contains the before and after values
  // of the changed row. The schema for the payload contains all the
  // columns that the change stream has tracked since the connector start
  // time.
  "schema": { 
    "type": "struct",
    "fields": [
      {
        // The schema for the before values of the changed row.
        "type": "struct",
        "fields": [
            {
                "type": "int32",
                "optional": false,
                "field": "id"
            },
            {
                "type": "string",
                "optional": true,
                "field": "first_name"
            }
        ],
        "optional": true,
        "name": "customers.Value",
        "field": "before"
      },
      {
        // The schema for the after values of the changed row.
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          }
        ],
          "optional": true,
          "name": "customers.Value",
          "field": "after"
        },
        {
          // The schema for the source metadata for the event.
          "type": "struct",
          "fields": [
            {
                "type": "string",
                "optional": false,
                "field": "version"
            },
            {
                "type": "string",
                "optional": false,
                "field": "connector"
            },
            {
                "type": "string",
                "optional": false,
                "field": "name"
            },
            {
                "type": "int64",
                "optional": false,
                "field": "ts_ms"
            },
            {
                "type": "boolean",
                "optional": true,
                "default": false,
                "field": "snapshot"
            },
            {
                "type": "string",
                "optional": false,
                "field": "db"
            },
            {
                "type": "string",
                "optional": false,
                "field": "sequence"
            },
            {
                "type": "string",
                "optional": false,
                "field": "project_id"
            },
            {
                "type": "string",
                "optional": false,
                "field": "instance_id"
            },
            {
                "type": "string",
                "optional": false,
                "field": "database_id"
            },
            {
                "type": "string",
                "optional": false,
                "field": "change_stream_name"
            },
            {
                "type": "string",
                "optional": true,
                "field": "table"
            }
            {
                "type": "string",
                "optional": true,
                "field": "server_transaction_id"
            }
            {
                "type": "int64",
                "optional": true,
                "field": "low_watermark"
            }
            {
                "type": "int64",
                "optional": true,
                "field": "read_at_timestamp"
            }
            {
                "type": "int64",
                "optional": true,
                "field": "number_of_records_in_transaction"
            }
            {
                "type": "string",
                "optional": true,
                "field": "transaction_tag"
            }
            {
                "type": "boolean",
                "optional": true,
                "field": "system_transaction"
            }
            {
                "type": "string",
                "optional": true,
                "field": "value_capture_type"
            }
            {
                "type": "string",
                "optional": true,
                "field": "partition_token"
            }
            {
                "type": "int32",
                "optional": true,
                "field": "mod_number"
            }
            {
                "type": "boolean",
                "optional": true,
                "field": "is_last_record_in_transaction_in_partition"
            }
            {
                "type": "int64",
                "optional": true,
                "field": "number_of_partitions_in_transaction"
            }
          ],
          "optional": false,
          "name": "io.debezium.connector.spanner.Source",
          "field": "source"
        },
      ]
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "connector_name.customers.Envelope"
  },
  "payload": {
    // The values of the row before the event.
    "before": null,
    // The values of the row after the event.
    "after": { 
        "id": 1,
        "first_name": "Anne",
    }
  },
  // The source metadata.
  "source": {
    "version": "{debezium-version}",
    "connector": "spanner",
    "name": "spanner_connector",
    "ts_ms": 1670955531785,
    "snapshot": "false",
    "db": "database",
    "sequence": "1",
    "project_id": "project",
    "instance_id": "instance",
    "database_id": "database",
    "change_stream_name": "change_stream",
    "table": "customers",
    "server_transaction_id": "transaction_id",
    "low_watermark": 1670955471635,
    "read_at_timestamp": 1670955531791,
    "number_records_in_transaction": 2,
    "transaction_tag": "",
    "system_transaction": false,
    "value_capture_type": "OLD_AND_NEW_VALUES",
    "partition_token": "partition_token",
    "mod_number": 0,
    "is_last_record_in_transaction_in_partition": true,
    "number_of_partitions_in_transaction": 1
  },
  "op": "c", 
  "ts_ms": 1559033904863 //
}

סימן מים נמוך

הסימן הנמוך מתאר את הזמן T שבו מובטח שמחבר Kafka העביר בסטרימינג ופרסם בנושא Kafka את כל האירועים עם חותמת זמן < T.

אפשר להפעיל את סימן המים הנמוך במחבר Kafka באמצעות הפרמטר gcp.spanner.low-watermark.enabled. הפרמטר הזה מושבת כברירת מחדל. אם סימן המים הנמוך מופעל, השדה low_watermark ברשומה של שינוי בנתוני זרם השינויים מאוכלס בחותמת הזמן של סימן המים הנמוך הנוכחי של מחבר Kafka.

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

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

נושאים שקשורים למטא-נתונים

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

אלה נושאי המטא-נתונים:

‫_consumer_offsets: נושא שנוצר אוטומטית על ידי Kafka. מאחסן את ההיסטים של הצרכנים עבור צרכנים שנוצרו במחבר Kafka.
‫_kafka-connect-offsets: נושא שנוצר באופן אוטומטי על ידי Kafka Connect. מאחסן את ההיסטים של המחבר.
‫_sync_topic_spanner_connector_connectorname: נושא שנוצר באופן אוטומטי על ידי המחבר. אחסון מטא-נתונים לגבי מחיצות של זרם שינויים.
‫_rebalancing_topic_spanner_connector_connectorname: נושא שנוצר באופן אוטומטי על ידי המחבר. המאפיין משמש לקביעת סטטוס הפעילות של משימת המחבר.
‫_debezium-heartbeat.connectorname: נושא שמשמש לעיבוד פעימות לב של שינוי בסטרימינג ב-Spanner.

זמן ריצה של מחבר Kafka

בהמשך מוסבר על זמן הריצה של מחבר Kafka.

מדרגיות

מחבר Kafka ניתן להרחבה אופקית, והוא פועל על משימה אחת או יותר שמתפרסות בין כמה עובדים של Kafka Connect.

התחייבויות מסירה של הודעות

מחבר Kafka תומך בהבטחת מסירה לפחות פעם אחת.

עמידות בפני תקלות

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

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

הגדרת מחבר Kafka

יצירת שידור חי של שינויים

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

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

התקנת קובץ ה-JAR של מחבר Kafka

אחרי שמתקינים את Zookeeper,‏ Kafka ו-Kafka Connect, כדי לפרוס מחבר Kafka צריך להוריד את ארכיון התוסף של המחבר, לחלץ את קובצי ה-JAR לסביבת Kafka Connect ולהוסיף את הספרייה עם קובצי ה-JAR אל Kafka Connect plugin.path. לאחר מכן צריך להפעיל מחדש את התהליך של Kafka Connect כדי שהמערכת תזהה את קובצי ה-JAR החדשים.

אם אתם עובדים עם קונטיינרים שלא ניתן לשנות, אתם יכולים לשלוף תמונות מתמונות הקונטיינר של Debezium עבור Zookeeper, ‏ Kafka ו-Kafka Connect. התמונה של Kafka Connect כוללת את מחבר Spanner שמותקן מראש.

למידע נוסף על התקנת קובצי JAR של מחבר Kafka שמבוססים על Debezium, אפשר לעיין במאמר בנושא התקנת Debezium.

הגדרת המחבר של Kafka

הדוגמה הבאה מציגה את ההגדרה של מחבר Kafka שמתחבר למקור נתונים לשינויים בשם changeStreamAll במסד הנתונים users במופע test-instance ובפרויקט test-project.

"name": "spanner-connector",
"config": {
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "gcp.spanner.project.id": "test-project",
    "gcp.spanner.instance.id": "test-instance",
    "gcp.spanner.database.id": "users",
    "gcp.spanner.change.stream": "changeStreamAll",
    "gcp.spanner.credentials.json": "{"client_id": user@example.com}",
    "gcp.spanner.database.role": "cdc-role",
    "tasks.max": "10"
}

ההגדרה הזו כוללת את הפריטים הבאים:

השם של מחבר כשרושמים אותו בשירות Kafka Connect.
השם של מחלקת המחברים הזו של Spanner.
מזהה הפרויקט.
מזהה מכונת Spanner.
המזהה של מסד הנתונים ב-Spanner.
השם של שידור השינויים.
אובייקט ה-JSON של מפתח חשבון השירות.
(אופציונלי) תפקיד במסד הנתונים של Spanner שבו רוצים להשתמש.
מספר המשימות המקסימלי.

רשימה מלאה של מאפייני המחבר זמינה במאמר מאפייני ההגדרה של מחבר Kafka.

הוספת הגדרות המחבר ל-Kafka Connect

כדי להתחיל להפעיל מחבר Spanner:

יוצרים הגדרה למחבר Spanner.
משתמשים ב-Kafka Connect REST API כדי להוסיף את הגדרות המחבר האלה לאשכול Kafka Connect.

אפשר לשלוח את ההגדרה הזו באמצעות הפקודה POST לשירות Kafka Connect פעיל. כברירת מחדל, שירות Kafka Connect פועל ביציאה 8083. השירות מתעד את ההגדרה ומתחיל את משימת המחבר שמתחבר למסד הנתונים של Spanner ומזרים רשומות של אירועי שינוי לנושאים של Kafka.

דוגמה לפקודה POST:

POST /connectors HTTP/1.1
Host: http://localhost:8083
Accept: application/json
{
  "name": "spanner-connector"
  "config": {
      "connector.class": "io.debezium.connector.spanner.SpannerConnector",
      "gcp.spanner.project.id": "test-project",
      "gcp.spanner.instance.id": "test-instance",
      "gcp.spanner.database.id": "users",
      "gcp.spanner.change.stream": "changeStreamAll",
      "gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
      "heartbeat.interval.ms": "100",
      "tasks.max": "10"
  }
}

דוגמה לתגובה מוצלחת:

HTTP/1.1 201 Created
Content-Type: application/json
{
    "name": "spanner-connector",
    "config": {
        "connector.class": "io.debezium.connector.spanner.SpannerConnector",
        "gcp.spanner.project.id": "test-project",
        "gcp.spanner.instance.id": "test-instance",
        "gcp.spanner.database.id": "users",
        "gcp.spanner.change.stream": "changeStreamAll",
        "gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
        "heartbeat.interval.ms": "100",
        "tasks.max": "10"
    },
    "tasks": [
        { "connector": "spanner-connector", "task": 1 },
        { "connector": "spanner-connector", "task": 2 },
        { "connector": "spanner-connector", "task": 3 }
    ]
}

עדכון ההגדרה של מחבר Kafka

כדי לעדכן את הגדרות המחבר, שולחים פקודה PUT לשירות Kafka Connect הפועל עם אותו שם מחבר.

נניח שיש לנו מחבר שפועל עם ההגדרה מהקטע הקודם. דוגמה לפקודה PUT:

PUT /connectors/spanner-connector/config HTTP/1.1
Host: http://localhost:8083
Accept: application/json
{
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "gcp.spanner.project.id": "test-project",
    "gcp.spanner.instance.id": "test-instance",
    "gcp.spanner.database.id": "users",
    "gcp.spanner.change.stream": "changeStreamAll",
    "gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
    "heartbeat.interval.ms": "100",
    "tasks.max": "10"
}

דוגמה לתגובה מוצלחת:

HTTP/1.1 200 OK
Content-Type: application/json
{
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "tasks.max": "10",
    "gcp.spanner.project.id": "test-project",
    "gcp.spanner.instance.id": "test-instance",
    "gcp.spanner.database.id": "users",
    "gcp.spanner.change.stream": "changeStreamAll",
    "gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
    "heartbeat.interval.ms": "100",
    "tasks.max": "10"
}

הפסקת המחבר של Kafka

כדי להפסיק את המחבר, שולחים פקודה DELETE לשירות Kafka Connect הפועל עם אותו שם מחבר.

נניח שיש לנו מחבר שפועל עם ההגדרה מהקטע הקודם. דוגמה לפקודה DELETE:

DELETE /connectors/spanner-connector HTTP/1.1
Host: http://localhost:8083

דוגמה לתגובה מוצלחת:

HTTP/1.1 204 No Content

מעקב אחרי מחבר Kafka

בנוסף למדדים הרגילים של Kafka Connect ושל Debezium, מחבר Kafka מייצא מדדים משלו:

‫MilliSecondsLowWatermark: ערך הסף התחתון הנוכחי של משימת המחבר באלפיות השנייה. הסימן הנמוך מתאר את הזמן T שבו מובטח שהמחבר העביר את כל האירועים עם חותמת זמן < T
‫MilliSecondsLowWatermarkLag: השהייה של סימן המים הנמוך ביחס לזמן הנוכחי באלפיות השנייה. כל האירועים עם חותמת זמן < T הועברו בסטרימינג
‫LatencyLowWatermark<Variant>MilliSeconds: השהייה של סימן המים הנמוך ביחס לזמן הנוכחי באלפיות השנייה. מוצגות וריאציות של P50,‏ P95,‏ P99,‏ Average,‏ Min ו-Max.
‫LatencySpanner<Variant>MilliSeconds: זמן האחזור של Spanner-commit-timestamp-to-connector-read. מוצגות וריאציות של P50,‏ P95,‏ P99,‏ Average,‏ Min ו-Max.
‫LatencyReadToEmit<Variant>MilliSeconds: זמן האחזור של Spanner-read-timestamp-to-connector-emit. מוצגות וריאציות של P50,‏ P95,‏ P99,‏ Average,‏ Min ו-Max.
‫LatencyCommitToEmit<Variant>tMilliSeconds: זמן האחזור של Spanner-commit-timestamp-to-connector-emit. מוצגות וריאציות של P50,‏ P95,‏ P99,‏ Average,‏ Min ו-Max.
‫LatencyCommitToPublish<Variant>MilliSeconds: זמן האחזור של חותמת הזמן של ביצוע הפעולה ב-Spanner עד לחותמת הזמן של הפרסום ב-Kafka. מוצגות וריאציות של P50,‏ P95,‏ P99,‏ Average,‏ Min ו-Max.
NumberOfChangeStreamPartitionsDetected: המספר הכולל של המחיצות שזוהו על ידי משימת המחבר הנוכחית.
‫NumberOfChangeStreamQueriesIssued: המספר הכולל של שאילתות של זרם השינויים שהונפקו על ידי המשימה הנוכחית.
‫NumberOfActiveChangeStreamQueries: המספר הפעיל של שאילתות של זרם שינויים שזוהו על ידי משימת המחבר הנוכחית.
‫SpannerEventQueueCapacity: הקיבולת הכוללת של StreamEventQueue, תור שמאחסן רכיבים שהתקבלו משאילתות של מקור נתונים לשינויים.
‫SpannerEventQueueCapacity: הקיבולת שנותרה.StreamEventQueue
‫TaskStateChangeEventQueueCapacity: הקיבולת הכוללת של TaskStateChangeEventQueue, תור שבו מאוחסנים אירועים שמתרחשים במחבר.
‫RemainingTaskStateChangeEventQueueCapacity: הקיבולת שנותרה.TaskStateChangeEventQueue
‫NumberOfActiveChangeStreamQueries: המספר הפעיל של שאילתות של זרם שינויים שזוהו על ידי משימת המחבר הנוכחית.

מאפייני ההגדרה של מחבר Kafka

אלה מאפייני החובה של התצורה של המחבר:

‫name: שם ייחודי של המחבר. ניסיון לרשום שוב את אותו שם יגרום לכשל. המאפיין הזה נדרש בכל המחברים של Kafka Connect.
‫connector.class: השם של מחלקת Java עבור המחבר. תמיד משתמשים בערך io.debezium.connector.spanner.SpannerConnector עבור מחבר Kafka.
‫tasks.max: המספר המקסימלי של המשימות שצריך ליצור עבור המחבר הזה.
‫gcp.spanner.project.id: מזהה הפרויקט
‫gcp.spanner.instance.id: מזהה מכונת Spanner
‫gcp.spanner.database.id: מזהה מסד הנתונים של Spanner
‫gcp.spanner.change.stream: השם של זרם השינויים ב-Spanner
‫gcp.spanner.credentials.json: אובייקט JSON של מפתח חשבון השירות.
‫gcp.spanner.credentials.path: נתיב הקובץ לאובייקט ה-JSON של מפתח חשבון השירות. חובה אם לא צוין הערך בשדה שלמעלה.
‫gcp.spanner.database.role : התפקיד במסד הנתונים של Spanner שבו רוצים להשתמש. הפעולה הזו נדרשת רק אם זרם השינויים מאובטח באמצעות בקרת גישה מפורטת. לתפקיד במסד הנתונים צריכה להיות הרשאת SELECT בפיד השינויים והרשאת EXECUTE בפונקציית הקריאה של פיד השינויים. מידע נוסף זמין במאמר בנושא בקרת גישה ברמת גרנולריות גבוהה לשינויים בנתונים.

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

‫gcp.spanner.low-watermark.enabled: מציין אם סימן המים הנמוך מופעל עבור המחבר. ברירת המחדל היא FALSE.
‫gcp.spanner.low-watermark.update-period.ms: המרווח שבו מתעדכן סימן המים הנמוך. ברירת המחדל היא 1,000 אלפיות השנייה.
‫heartbeat.interval.ms: האינטרוול של אותות החיים ב-Spanner. ברירת המחדל היא 300,000 (חמש דקות).
‫gcp.spanner.start.time: שעת ההתחלה של המחבר. ברירת המחדל היא השעה הנוכחית.
‫gcp.spanner.end.time: שעת הסיום של המחבר. ברירת המחדל היא אינסוף.
‫tables.exclude.list: הטבלאות שרוצים להחריג מהאירועים של שינויים. ברירת המחדל היא ריק.
‫tables.include.list: הטבלאות שרוצים לכלול לגביהן אירועי שינוי. אם לא תציינו טבלאות, כל הטבלאות ייכללו. ברירת המחדל היא ריק.
‫gcp.spanner.stream.event.queue.capacity: הקיבולת של תור האירועים ב-Spanner. ברירת המחדל היא 10,000.
‫connector.spanner.task.state.change.event.queue.capacity: הקיבולת של תור האירועים של שינוי מצב המשימה. ברירת המחדל היא 1,000.
‫connector.spanner.max.missed.heartbeats: המספר המקסימלי של פעימות לב שהוחמצו בשאילתת שינוי נתונים לפני שמוחזר חריג. ברירת המחדל היא 10.
‫scaler.monitor.enabled: מציין אם ההתאמה האוטומטית לעומס של המשימות מופעלת. ברירת המחדל היא False.
‫tasks.desired.partitions: מספר המחיצות המועדף של סנכרון שינויים בזרמי נתונים לכל משימה. הפרמטר הזה נדרש לצורך התאמה אוטומטית לעומס של המשימה. ברירת המחדל היא 2.
‫tasks.min: המספר המינימלי של המשימות. הפרמטר הזה נדרש לצורך התאמה אוטומטית לעומס של המשימה. ברירת המחדל היא 1.
‫connector.spanner.sync.topic: השם של נושא הסנכרון, נושא פנימי של מחבר שמשמש לאחסון תקשורת בין משימות. אם המשתמש לא סיפק שם, ערך ברירת המחדל הוא _sync_topic_spanner_connector_connectorname.
‫connector.spanner.sync.poll.duration: משך הזמן להמתנה בין בדיקות של נושא הסנכרון. ברירת המחדל היא 500 אלפיות השנייה.
‫connector.spanner.sync.request.timeout.ms: הזמן הקצוב לתפוגה של בקשות לנושא הסנכרון. ברירת המחדל היא 5,000 אלפיות השנייה.
‫connector.spanner.sync.delivery.timeout.ms: הזמן הקצוב לתפוגה של פרסום בנושא הסנכרון. ברירת המחדל היא 15,000 אלפיות השנייה.
‫connector.spanner.sync.commit.offsets.interval.ms: מרווח הזמן שבו מתבצעת הקצאה של היסטים לנושא הסנכרון. ברירת המחדל היא 60,000 אלפיות השנייה.
‫connector.spanner.sync.publisher.wait.timeout: המרווח שבו ההודעות מתפרסמות בנושא הסנכרון. ברירת המחדל היא 5 אלפיות השנייה.
‫connector.spanner.rebalancing.topic: השם של נושא האיזון מחדש. הנושא rebalancing הוא נושא פנימי של מחבר שמשמש לקביעת סטטוס הפעילות של המשימה. אם המשתמש לא סיפק שם, ערך ברירת המחדל הוא _rebalancing_topic_spanner_connector_connectorname.
‫connector.spanner.rebalancing.poll.duration: משך הזמן של הסקר בנושא איזון מחדש. ברירת המחדל היא 5,000 אלפיות השנייה.
‫connector.spanner.rebalancing.commit.offsets.timeout: הזמן הקצוב לתפוגה של ביצוע אופסטים (offset) לנושא האיזון מחדש. ברירת המחדל היא 5,000 אלפיות השנייה.
‫connector.spanner.rebalancing.commit.offsets.interval.ms: מרווח הזמן שבו מתבצעת הקצאה של היסטים לנושא הסנכרון. ברירת המחדל היא 60,000 אלפיות השנייה.
‫connector.spanner.rebalancing.task.waiting.timeout: משך הזמן שבו משימה ממתינה לפני עיבוד אירוע איזון מחדש. ברירת המחדל היא 1,000 אלפיות השנייה.

רשימה מפורטת יותר של מאפייני המחבר שניתנים להגדרה זמינה במאגר GitHub.

מגבלות

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

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

מושגי ליבה

Debezium

מחבר Kafka

פלט של מחבר Kafka

הקלטת הזמנה

שינוי אירועים

סימן מים נמוך

נושאים שקשורים למטא-נתונים

זמן ריצה של מחבר Kafka

מדרגיות

התחייבויות מסירה של הודעות

עמידות בפני תקלות

הגדרת מחבר Kafka

יצירת שידור חי של שינויים

התקנת קובץ ה-JAR של מחבר Kafka

הגדרת המחבר של Kafka

הוספת הגדרות המחבר ל-Kafka Connect

עדכון ההגדרה של מחבר Kafka

הפסקת המחבר של Kafka

מעקב אחרי מחבר Kafka

מאפייני ההגדרה של מחבר Kafka

מגבלות

יצירת חיבורים של סנכרון שינויים בזרמי נתונים אל Kafka