חותמות זמן מפוצלות

אם אוסף מכיל מסמכים עם ערכים רציפים באינדקס, Cloud Firestore מגביל את קצב הכתיבה ל-500 כתיבות בשנייה. בדף הזה מוסבר איך לפצל שדה במסמך כדי לעקוף את המגבלה הזו. קודם כל, נגדיר מה הכוונה ב'שדות רציפים באינדקס' ונבהיר מתי חלה ההגבלה הזו.

שדות באינדקס ברצף

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

לדוגמה, המגבלה חלה על אוסף של user מסמכים עם שדה מאונדקס userid אם האפליקציה מקצה ערכים של userid באופן הבא:

  • 1281, 1282, 1283, 1284, 1285, ...

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

  • 100000, 100001, 100002, 100003, ...
  • 0, 1, 2, 3, ...

חלוקה לשברי נתונים של שדה חותמת זמן

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

  1. מוסיפים שדה shard לצד השדה timestamp. בשדה shard צריך להשתמש ב-1..n ערכים שונים. כך מגדילים את מגבלת הכתיבה לאוסף ל-500*n, אבל צריך לצבור n שאילתות.
  2. מעדכנים את לוגיקת הכתיבה כך שיוקצה ערך shard לכל מסמך באופן אקראי.
  3. צריך לעדכן את השאילתות כדי לצבור את קבוצות התוצאות המפוצלות.
  4. משביתים את האינדקסים של שדה יחיד גם בשדה shard וגם בשדה timestamp. מוחקים אינדקסים מורכבים קיימים שמכילים את השדה timestamp.
  5. יוצרים אינדקסים מורכבים חדשים כדי לתמוך בשאילתות המעודכנות. הסדר של השדות באינדקס חשוב, והשדה shard חייב להופיע לפני השדה timestamp. כל האינדקסים שכוללים את השדה timestamp חייבים לכלול גם את השדה shard.

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

בדוגמאות הבאות מוסבר איך לפצל שדה timestamp ואיך לשלוח שאילתה למערך תוצאות מפוצל.

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

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

Node.js
async function insertData() {
  const instruments = [
    {
      symbol: 'AAA',
      price: {
        currency: 'USD',
        micros: 34790000
      },
      exchange: 'EXCHG1',
      instrumentType: 'commonstock',
      timestamp: Timestamp.fromMillis(
          Date.parse('2019-01-01T13:45:23.010Z'))
    },
    {
      symbol: 'BBB',
      price: {
        currency: 'JPY',
        micros: 64272000000
      },
      exchange: 'EXCHG2',
      instrumentType: 'commonstock',
      timestamp: Timestamp.fromMillis(
          Date.parse('2019-01-01T13:45:23.101Z'))
    },
    {
      symbol: 'Index1 ETF',
      price: {
        currency: 'USD',
        micros: 473000000
      },
      exchange: 'EXCHG1',
      instrumentType: 'etf',
      timestamp: Timestamp.fromMillis(
          Date.parse('2019-01-01T13:45:23.001Z'))
    }
  ];

  const batch = fs.batch();
  for (const inst of instruments) {
    const ref = fs.collection('instruments').doc();
    batch.set(ref, inst);
  }

  await batch.commit();
}

nonShardedTimestamps.js

האפליקציה הזו מריצה את השאילתות הבאות וממיינת לפי השדה timestamp:

Node.js
function createQuery(fieldName, fieldOperator, fieldValue, limit = 5) {
  return fs.collection('instruments')
      .where(fieldName, fieldOperator, fieldValue)
      .orderBy('timestamp', 'desc')
      .limit(limit)
      .get();
}

function queryCommonStock() {
  return createQuery('instrumentType', '==', 'commonstock');
}

function queryExchange1Instruments() {
  return createQuery('exchange', '==', 'EXCHG1');
}

function queryUSDInstruments() {
  return createQuery('price.currency', '==', 'USD');
}

nonShardedTimestamps.js
 
insertData()
    .then(() => {
      const commonStock = queryCommonStock()
          .then(
              (docs) => {
                console.log('--- queryCommonStock: ');
                docs.forEach((doc) => {
                  console.log(`doc = ${util.inspect(doc.data(), {depth: 4})}`);
                });
              }
          );
      const exchange1Instruments = queryExchange1Instruments()
          .then(
              (docs) => {
                console.log('--- queryExchange1Instruments: ');
                docs.forEach((doc) => {
                  console.log(`doc = ${util.inspect(doc.data(), {depth: 4})}`);
                });
              }
          );
      const usdInstruments = queryUSDInstruments()
          .then(
              (docs) => {
                console.log('--- queryUSDInstruments: ');
                docs.forEach((doc) => {
                  console.log(`doc = ${util.inspect(doc.data(), {depth: 4})}`);
                });
              }
          );
      return Promise.all([commonStock, exchange1Instruments, usdInstruments]);
    });
nonShardedTimestamps.js

אחרי מחקר מסוים, אתם מגיעים למסקנה שהאפליקציה תקבל בין 1,000 ל-1,500 עדכונים של מכשירים בשנייה. המספר הזה גבוה מהמגבלה של 500 פעולות כתיבה בשנייה שמותרות לאוספים שמכילים מסמכים עם שדות של חותמות זמן שעברו אינדוקס. כדי להגדיל את קצב העברת הנתונים לכתיבה, צריך 3 ערכי שבר, MAX_INSTRUMENT_UPDATES/500 = 3. בדוגמה הזו נעשה שימוש בערכי השבר x, y ו-z. אפשר להשתמש גם במספרים או בתווים אחרים לערכי השבר.

הוספת שדה של שברי נתונים

מוסיפים שדה shard למסמכים. מגדירים את השדה shard לערכים x, y או z, וכך מגדילים את מגבלת הכתיבה באוסף ל-1,500 פעולות כתיבה בשנייה.

Node.js
// Define our 'K' shard values
const shards = ['x', 'y', 'z'];
// Define a function to help 'chunk' our shards for use in queries.
// When using the 'in' query filter there is a max number of values that can be
// included in the value. If our number of shards is higher than that limit
// break down the shards into the fewest possible number of chunks.
function shardChunks() {
  const chunks = [];
  let start = 0;
  while (start < shards.length) {
    const elements = Math.min(MAX_IN_VALUES, shards.length - start);
    const end = start + elements;
    chunks.push(shards.slice(start, end));
    start = end;
  }
  return chunks;
}

// Add a convenience function to select a random shard
function randomShard() {
  return shards[Math.floor(Math.random() * Math.floor(shards.length))];
}

shardedTimestamps.js
 
async function insertData() {
  const instruments = [
    {
      shard: randomShard(),  // add the new shard field to the document
      symbol: 'AAA',
      price: {
        currency: 'USD',
        micros: 34790000
      },
      exchange: 'EXCHG1',
      instrumentType: 'commonstock',
      timestamp: Timestamp.fromMillis(
          Date.parse('2019-01-01T13:45:23.010Z'))
    },
    {
      shard: randomShard(),  // add the new shard field to the document
      symbol: 'BBB',
      price: {
        currency: 'JPY',
        micros: 64272000000
      },
      exchange: 'EXCHG2',
      instrumentType: 'commonstock',
      timestamp: Timestamp.fromMillis(
          Date.parse('2019-01-01T13:45:23.101Z'))
    },
    {
      shard: randomShard(),  // add the new shard field to the document
      symbol: 'Index1 ETF',
      price: {
        currency: 'USD',
        micros: 473000000
      },
      exchange: 'EXCHG1',
      instrumentType: 'etf',
      timestamp: Timestamp.fromMillis(
          Date.parse('2019-01-01T13:45:23.001Z'))
    }
  ];

  const batch = fs.batch();
  for (const inst of instruments) {
    const ref = fs.collection('instruments').doc();
    batch.set(ref, inst);
  }

  await batch.commit();
}

shardedTimestamps.js

שאילתה של חותמת הזמן המפוצלת

כדי להוסיף שדה shard, צריך לעדכן את השאילתות כדי לצבור תוצאות מפוצלות:

Node.js
function createQuery(fieldName, fieldOperator, fieldValue, limit = 5) {
  // For each shard value, map it to a new query which adds an additional
  // where clause specifying the shard value.
  return Promise.all(shardChunks().map(shardChunk => {
        return fs.collection('instruments')
            .where('shard', 'in', shardChunk)  // new shard condition
            .where(fieldName, fieldOperator, fieldValue)
            .orderBy('timestamp', 'desc')
            .limit(limit)
            .get();
      }))
      // Now that we have a promise of multiple possible query results, we need
      // to merge the results from all of the queries into a single result set.
      .then((snapshots) => {
        // Create a new container for 'all' results
        const docs = [];
        snapshots.forEach((querySnapshot) => {
          querySnapshot.forEach((doc) => {
            // append each document to the new all container
            docs.push(doc);
          });
        });
        if (snapshots.length === 1) {
          // if only a single query was returned skip manual sorting as it is
          // taken care of by the backend.
          return docs;
        } else {
          // When multiple query results are returned we need to sort the
          // results after they have been concatenated.
          // 
          // since we're wanting the `limit` newest values, sort the array
          // descending and take the first `limit` values. By returning negated
          // values we can easily get a descending value.
          docs.sort((a, b) => {
            const aT = a.data().timestamp;
            const bT = b.data().timestamp;
            const secondsDiff = aT.seconds - bT.seconds;
            if (secondsDiff === 0) {
              return -(aT.nanoseconds - bT.nanoseconds);
            } else {
              return -secondsDiff;
            }
          });
          return docs.slice(0, limit);
        }
      });
}

function queryCommonStock() {
  return createQuery('instrumentType', '==', 'commonstock');
}

function queryExchange1Instruments() {
  return createQuery('exchange', '==', 'EXCHG1');
}

function queryUSDInstruments() {
  return createQuery('price.currency', '==', 'USD');
}

shardedTimestamps.js
 
insertData()
    .then(() => {
      const commonStock = queryCommonStock()
          .then(
              (docs) => {
                console.log('--- queryCommonStock: ');
                docs.forEach((doc) => {
                  console.log(`doc = ${util.inspect(doc.data(), {depth: 4})}`);
                });
              }
          );
      const exchange1Instruments = queryExchange1Instruments()
          .then(
              (docs) => {
                console.log('--- queryExchange1Instruments: ');
                docs.forEach((doc) => {
                  console.log(`doc = ${util.inspect(doc.data(), {depth: 4})}`);
                });
              }
          );
      const usdInstruments = queryUSDInstruments()
          .then(
              (docs) => {
                console.log('--- queryUSDInstruments: ');
                docs.forEach((doc) => {
                  console.log(`doc = ${util.inspect(doc.data(), {depth: 4})}`);
                });
              }
          );
      return Promise.all([commonStock, exchange1Instruments, usdInstruments]);
    });
shardedTimestamps.js

עדכון הגדרות האינדקס

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

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

מסוף Firebase

  1. פותחים את הדף Cloud Firestore Composite Indexes במסוף Firebase.

    כניסה לדף Composite Indexes

  2. לכל אינדקס שמכיל את השדה timestamp, לוחצים על הלחצן ואז על מחיקה.

קונסולת GCP

  1. נכנסים לדף Databases במסוף Google Cloud.

    כניסה לדף Databases

  2. בוחרים את מסד הנתונים הרצוי מרשימת מסדי הנתונים.

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

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

  5. לכל אחד מהאינדקסים האלה, לוחצים על הלחצן ואז על מחיקה.

Firebase CLI

  1. אם לא הגדרתם את Firebase CLI, פועלים לפי ההוראות האלה כדי להתקין את ה-CLI ולהריץ את הפקודה firebase init. במהלך הפקודה init, חשוב לבחור באפשרות Firestore: Deploy rules and create indexes for Firestore.
  2. במהלך ההגדרה, Firebase CLI מוריד את הגדרות האינדקס הקיימות לקובץ שנקרא, כברירת מחדל, firestore.indexes.json.

  3. מסירים את כל הגדרות האינדקס שמכילות את השדה timestamp, לדוגמה:

    {
"indexes": [
  // Delete composite index definition that contain the timestamp field
  {
    "collectionGroup": "instruments",
    "queryScope": "COLLECTION",
    "fields": [
      {
        "fieldPath": "exchange",
        "order": "ASCENDING"
      },
      {
        "fieldPath": "timestamp",
        "order": "DESCENDING"
      }
    ]
  },
  {
    "collectionGroup": "instruments",
    "queryScope": "COLLECTION",
    "fields": [
      {
        "fieldPath": "instrumentType",
        "order": "ASCENDING"
      },
      {
        "fieldPath": "timestamp",
        "order": "DESCENDING"
      }
    ]
  },
  {
    "collectionGroup": "instruments",
    "queryScope": "COLLECTION",
    "fields": [
      {
        "fieldPath": "price.currency",
        "order": "ASCENDING"
      },
      {
        "fieldPath": "timestamp",
        "order": "DESCENDING"
      }
    ]
  },
 ]
}

  4. פורסים את הגדרות האינדקס המעודכנות:

    firebase deploy --only firestore:indexes

עדכון של הגדרות אינדקס של שדה יחיד

מסוף Firebase

  1. פותחים את הדף Cloud Firestore Single Field Indexes במסוף Firebase.

    מעבר אל Single Field Indexes

  2. לוחצים על הוספת פטור.

  3. בשדה מזהה אוסף, מזינים instruments. בשדה נתיב השדה, מזינים timestamp.

  4. בקטע היקף השאילתה, בוחרים באפשרות אוסף וגם באפשרות קבוצת אוספים.

  5. לוחצים על הבא.

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

  7. חוזרים על אותם שלבים בשדה shard.

קונסולת GCP

  1. נכנסים לדף Databases במסוף Google Cloud.

    כניסה לדף Databases

  2. בוחרים את מסד הנתונים הרצוי מרשימת מסדי הנתונים.

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

  4. לוחצים על הכרטיסייה שדה יחיד.

  5. לוחצים על הוספת פטור.

  6. בשדה מזהה אוסף, מזינים instruments. בשדה נתיב השדה, מזינים timestamp.

  7. בקטע היקף השאילתה, בוחרים באפשרות אוסף וגם באפשרות קבוצת אוספים.

  8. לוחצים על הבא.

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

  10. חוזרים על אותם שלבים בשדה shard.

Firebase CLI

  1. מוסיפים את הטקסט הבא לקטע fieldOverrides בקובץ הגדרות האינדקס:

    {
 "fieldOverrides": [
   // Disable single-field indexing for the timestamp field
   {
     "collectionGroup": "instruments",
     "fieldPath": "timestamp",
     "indexes": []
   },
 ]
}

  2. פורסים את הגדרות האינדקס המעודכנות:

    firebase deploy --only firestore:indexes

יצירת אינדקסים מורכבים חדשים

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

איסוף שדות באינדקס היקף השאילתה
כלי נגינה ‫ shard, price.currency, timestamp איסוף
כלי נגינה shard, exchange, timestamp איסוף
כלי נגינה ‫ shard, ‏ instrumentType, ‏ timestamp איסוף

הודעות שגיאה

כדי ליצור את האינדקסים האלה, מריצים את השאילתות המעודכנות.

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

Firebase CLI

  1. מוסיפים את האינדקסים הבאים לקובץ הגדרת האינדקס:

     {
   "indexes": [
   // New indexes for sharded timestamps
     {
       "collectionGroup": "instruments",
       "queryScope": "COLLECTION",
       "fields": [
         {
           "fieldPath": "shard",
           "order": "DESCENDING"
         },
         {
           "fieldPath": "exchange",
           "order": "ASCENDING"
         },
         {
           "fieldPath": "timestamp",
           "order": "DESCENDING"
         }
       ]
     },
     {
       "collectionGroup": "instruments",
       "queryScope": "COLLECTION",
       "fields": [
         {
           "fieldPath": "shard",
           "order": "DESCENDING"
         },
         {
           "fieldPath": "instrumentType",
           "order": "ASCENDING"
         },
         {
           "fieldPath": "timestamp",
           "order": "DESCENDING"
         }
       ]
     },
     {
       "collectionGroup": "instruments",
       "queryScope": "COLLECTION",
       "fields": [
         {
           "fieldPath": "shard",
           "order": "DESCENDING"
         },
         {
           "fieldPath": "price.currency",
           "order": "ASCENDING"
         },
         {
           "fieldPath": "timestamp",
           "order": "DESCENDING"
         }
       ]
     },
   ]
 }

  2. פורסים את הגדרות האינדקס המעודכנות:

    firebase deploy --only firestore:indexes

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

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

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

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

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