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

אם אוסף מכיל מסמכים עם ערכים עוקבים שנוסף להם אינדקס, 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. צריך להשתמש ב-1..n ערכים נפרדים בשדה shard. כך מגדילים את מגבלת הכתיבה של האוסף ל-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();
}

האפליקציה הזו מריצה את השאילתות הבאות ומסדרת אותן לפי השדה 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');
}
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]);
    });

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

הוספת שדה של פיצול (shard)

מוסיפים שדה 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))];
}
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();
}

שליחת שאילתה לגבי חותמת הזמן של המקטע

אם מוסיפים שדה 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');
}
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]);
    });

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

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

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

מסוף Firebase

  1. במסוף Firebase, נכנסים לכרטיסייה Databases & Storage > Firestore > Composite Indexes.

    כניסה לדף Composite Indexes

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

קונסולת GCP

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

    כניסה לדף Databases

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

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

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

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

Firebase CLI

  1. אם עדיין לא הגדרתם את ה-CLI של Firebase, פעלו לפי ההוראות האלה כדי להתקין את ה-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. במסוף Firebase, נכנסים לכרטיסייה Databases & Storage > Firestore > Single Field Indexes.

    כניסה לדף Single Field Indexes

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

  3. בשדה מזהה האוסף, מזינים instruments. בשדה Field path מזינים את הערך timestamp.

  4. בקטע Query scope, בוחרים גם באפשרות Collection וגם באפשרות Collection group.

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

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

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

קונסולת GCP

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

    כניסה לדף Databases

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

  3. בתפריט הניווט, לוחצים על Indexes ואז על הכרטיסייה Single Field.

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

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

  6. בשדה מזהה האוסף, מזינים instruments. בשדה Field path מזינים את הערך timestamp.

  7. בקטע Query scope, בוחרים גם באפשרות Collection וגם באפשרות Collection group.

  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, ‏ instrumentType, ‏ חותמת זמן איסוף

הודעות שגיאה

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

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

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

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

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