Umgang mit Sharded-Daten in verteilten Systemen: Ein tiefer Einblick in Joins, Broadcasts und Abfrageoptimierung

In modernen verteilten Datenbanken hat die Notwendigkeit der horizontalen Skalierung von Daten zur weit verbreiteten Einführung von Sharding geführt. Während Sharding dabei hilft, große Datensätze über mehrere Knoten hinweg zu verwalten, bringt es Herausforderungen mit sich, insbesondere bei der Durchführung von Joins und der Gewährleistung eines effizienten Datenabrufs. In diesem Artikel untersuchen wir verschiedene Konzepte und Techniken, die diese Herausforderungen angehen, und konzentrieren uns dabei insbesondere auf Broadcast-Joins, Shard-Key-Alignment und verteilte Abfrage-Engines wie Presto und BigQuery. Darüber hinaus zeigen wir, wie diese Probleme in realen Anwendungen mit Node.js und Express gelöst werden können.

---

Sharding-Beispiel in Node.js mit Express.js

So können Sie Sharding in PostgreSQL mithilfe von Node.js und Express.js implementieren.

PostgreSQL-Sharding-Beispiel

Verwendung von Citus oder manuelles logisches Sharding mit Node.js:

Beispiel mit logischem Sharding

1. Einrichtungstabellen für Shards:
   Verwenden Sie Tabellen für Shards (user_data auf Shard1 und user_data auf Shard2).

2. Erstellen Sie eine Express.js-API:
   Verteilen Sie Abfragen basierend auf einem Shard-Schlüssel (z. B. Benutzer-ID).
   

   const express = require('express');
   const { Pool } = require('pg');

   const poolShard1 = new Pool({ connectionString: 'postgresql://localhost/shard1' });
   const poolShard2 = new Pool({ connectionString: 'postgresql://localhost/shard2' });

   const app = express();
   app.use(express.json());

   const getShardPool = (userId) => (userId % 2 === 0 ? poolShard1 : poolShard2);

   app.post('/user', async (req, res) => {
       const { userId, data } = req.body;
       const pool = getShardPool(userId);
       try {
           await pool.query('INSERT INTO user_data (user_id, data) VALUES (, )', [userId, data]);
           res.status(200).send('User added successfully');
       } catch (err) {
           console.error(err);
           res.status(500).send('Error inserting user');
       }
   });

   app.get('/user/:userId', async (req, res) => {
       const userId = parseInt(req.params.userId, 10);
       const pool = getShardPool(userId);
       try {
           const result = await pool.query('SELECT * FROM user_data WHERE user_id = ', [userId]);
           res.status(200).json(result.rows);
       } catch (err) {
           console.error(err);
           res.status(500).send('Error retrieving user');
       }
   });

   app.listen(3000, () => console.log('Server running on port 3000'));

---

1. Sharding in verteilten Datenbanken

Sharding ist der Prozess der horizontalen Partitionierung von Daten über mehrere Datenbankinstanzen oder Shards, um Leistung, Skalierbarkeit und Verfügbarkeit zu verbessern. Sharding ist häufig erforderlich, wenn eine einzelne Datenbankinstanz das Daten- oder Datenverkehrsvolumen nicht bewältigen kann.

Sharding-Strategien:

• Bereichsbasiertes Sharding: Daten werden basierend auf dem Bereich eines Schlüssels über Shards verteilt, z. B. Partitionierung von Bestellungen nach Bestelldatum.
• Hash-basiertes Sharding: Daten werden durch einen Shard-Schlüssel (z. B. Benutzer-ID) gehasht, um die Daten gleichmäßig auf Shards zu verteilen.
• Verzeichnisbasiertes Sharding: Ein zentrales Verzeichnis verfolgt, wo sich Daten im System befinden.

Wenn jedoch zusammengehörige Tabellen auf verschiedene Schlüssel aufgeteilt werden oder wenn eine Tabelle eine Verknüpfung mit einer anderen Tabelle über mehrere Shards erfordert, kann sich die Leistung aufgrund der Notwendigkeit von Scatter-Gather-Vorgängen verschlechtern. Hier ist das Verständnis von Broadcast-Joins und der Shard-Key-Ausrichtung von entscheidender Bedeutung.

---

2. Herausforderungen mit Joins in Sharded-Systemen

Wenn sich Daten in verschiedenen Shards befinden, kann die Durchführung von Verknüpfungen zwischen diesen Shards komplex sein. Hier ist eine Aufschlüsselung der häufigsten Herausforderungen:

1. Shard Key-Fehlausrichtung:

In vielen Systemen werden Tabellen auf verschiedene Schlüssel aufgeteilt. Zum Beispiel:

• Die Tabelle Benutzer ist möglicherweise nach Benutzer-ID fragmentiert.
• Die Tabelle Bestellungen kann nach Region aufgeteilt sein.

Beim Durchführen einer Verknüpfung (z. B. Bestellungen.Benutzer_ID = Benutzer.Benutzer_ID) muss das System Daten von mehreren Shards abrufen, da sich die relevanten Datensätze möglicherweise nicht im selben Shard befinden.

2. Scatter-Gather-Joins:

Bei einem Scatter-Gather-Join muss das System:

• Anfragen an alle Shards senden, die relevante Daten enthalten.
• Ergebnisse über Shards hinweg aggregieren. Dies kann die Leistung erheblich beeinträchtigen, insbesondere wenn die Daten über viele Shards verteilt sind.

3. Broadcast Joins:

Ein Broadcast-Join tritt auf, wenn eine der zu verbindenden Tabellen klein genug ist, um an alle Shards gesendet zu werden. In diesem Fall:

• Die kleine Tabelle (z. B. Benutzer) wird auf allen Knoten repliziert, auf denen sich die größere, fragmentierte Tabelle (z. B. Bestellungen) befindet.
• Jeder Knoten kann dann seine lokalen Daten mit den gesendeten Daten verbinden, wodurch die Notwendigkeit einer Shard-übergreifenden Kommunikation entfällt.

---

3. Verteilte Abfrage-Engines für Sharded-Daten verwenden

Verteilte Abfrage-Engines wie Presto und BigQuery sind darauf ausgelegt, fragmentierte Daten zu verarbeiten und Abfragen effizient über verteilte Systeme hinweg zusammenzuführen.

Presto/Trino:

Presto ist eine verteilte SQL-Abfrage-Engine, die für die Abfrage großer Datensätze aus heterogenen Datenquellen (z. B. relationale Datenbanken, NoSQL-Datenbanken, Data Lakes) entwickelt wurde. Presto führt Verknüpfungen über verteilte Datenquellen hinweg durch und kann Abfragen optimieren, indem es die Datenbewegung zwischen Knoten minimiert.

Beispielhafter Anwendungsfall: Shard-Daten mit Presto verbinden

In einem Szenario, in dem Bestellungen nach Region und Benutzer nach Benutzer-ID aufgeteilt werden, kann Presto mithilfe seines verteilten Ausführungsmodells einen Join über verschiedene Shards hinweg durchführen.

Abfrage:

   const express = require('express');
   const { Pool } = require('pg');

   const poolShard1 = new Pool({ connectionString: 'postgresql://localhost/shard1' });
   const poolShard2 = new Pool({ connectionString: 'postgresql://localhost/shard2' });

   const app = express();
   app.use(express.json());

   const getShardPool = (userId) => (userId % 2 === 0 ? poolShard1 : poolShard2);

   app.post('/user', async (req, res) => {
       const { userId, data } = req.body;
       const pool = getShardPool(userId);
       try {
           await pool.query('INSERT INTO user_data (user_id, data) VALUES (, )', [userId, data]);
           res.status(200).send('User added successfully');
       } catch (err) {
           console.error(err);
           res.status(500).send('Error inserting user');
       }
   });

   app.get('/user/:userId', async (req, res) => {
       const userId = parseInt(req.params.userId, 10);
       const pool = getShardPool(userId);
       try {
           const result = await pool.query('SELECT * FROM user_data WHERE user_id = ', [userId]);
           res.status(200).json(result.rows);
       } catch (err) {
           console.error(err);
           res.status(500).send('Error retrieving user');
       }
   });

   app.listen(3000, () => console.log('Server running on port 3000'));

Presto wird:

1. Verwenden Sie Scatter-Gather, um relevante Benutzerdatensätze abzurufen.
2. Daten knotenübergreifend zusammenführen.

Google BigQuery:

BigQuery ist ein vollständig verwaltetes, serverloses Data Warehouse, das sich hervorragend für die Ausführung umfangreicher analytischer Abfragen eignet. Während BigQuery die Details des Shardings abstrahiert, partitioniert und verteilt es Daten automatisch auf viele Knoten, um die Abfrage zu optimieren. Es kann große Datensätze problemlos verarbeiten und ist besonders effektiv für analytische Abfragen, bei denen Daten nach Zeit oder anderen Dimensionen partitioniert sind.

Beispielanwendungsfall: Verknüpfen von Sharded-Tabellen in BigQuery

   const express = require('express');
   const { Pool } = require('pg');

   const poolShard1 = new Pool({ connectionString: 'postgresql://localhost/shard1' });
   const poolShard2 = new Pool({ connectionString: 'postgresql://localhost/shard2' });

   const app = express();
   app.use(express.json());

   const getShardPool = (userId) => (userId % 2 === 0 ? poolShard1 : poolShard2);

   app.post('/user', async (req, res) => {
       const { userId, data } = req.body;
       const pool = getShardPool(userId);
       try {
           await pool.query('INSERT INTO user_data (user_id, data) VALUES (, )', [userId, data]);
           res.status(200).send('User added successfully');
       } catch (err) {
           console.error(err);
           res.status(500).send('Error inserting user');
       }
   });

   app.get('/user/:userId', async (req, res) => {
       const userId = parseInt(req.params.userId, 10);
       const pool = getShardPool(userId);
       try {
           const result = await pool.query('SELECT * FROM user_data WHERE user_id = ', [userId]);
           res.status(200).json(result.rows);
       } catch (err) {
           console.error(err);
           res.status(500).send('Error retrieving user');
       }
   });

   app.listen(3000, () => console.log('Server running on port 3000'));

BigQuery übernimmt automatisch die Partitionierung und Verteilung und minimiert so den Bedarf an manuellem Sharding.

---

4. Umgang mit der Fehlausrichtung von Shard-Schlüsseln in Node.js-Anwendungen

Beim Umgang mit Shard-Daten in Node.js-Anwendungen treten häufig Probleme wie falsch ausgerichtete Shard-Schlüssel und die Notwendigkeit von Scatter-Gather-Verbindungen auf. So können Sie diese Herausforderungen mit Node.js und Express angehen.

Verarbeitung von Broadcast-Joins in Node.js

Wenn ein Join die Übertragung einer kleinen Tabelle (z. B. Benutzer) über alle Shards erfordert, können Sie den Join in der Anwendungsschicht implementieren, indem Sie die kleine Tabelle einmal abrufen und sie zum Joinen mit Daten aus Shard-Tabellen verwenden.

SELECT o.order_id, u.user_name
FROM orders o
JOIN users u
ON o.user_id = u.user_id;

Verarbeitung von Scatter-Gather-Abfragen in Node.js

Bei Abfragen, die Scatter-Gather-Joins beinhalten (z. B. wenn Shard-Schlüssel falsch ausgerichtet sind), müssen Sie alle Shards abfragen und die Ergebnisse in Ihrer Anwendungsschicht aggregieren.

SELECT o.order_id, u.user_name
FROM `project.dataset.orders` o
JOIN `project.dataset.users` u
ON o.user_id = u.user_id
WHERE o.order_date BETWEEN '2024-01-01' AND '2024-12-31';

---

5. Best Practices für die Abfrageoptimierung mit Sharded Data

Berücksichtigen Sie beim Umgang mit Sharded-Daten und der Durchführung von Verknüpfungen die folgenden Best Practices:

1. Shard-Schlüssel ausrichten: Stellen Sie nach Möglichkeit sicher, dass verwandte Tabellen denselben Shard-Schlüssel verwenden. Dies minimiert die Notwendigkeit von Cross-Shard-Joins und verbessert die Leistung.

2. Denormalisierung: In Szenarien, in denen Verknüpfungen häufig vorkommen, sollten Sie eine Denormalisierung Ihrer Daten in Betracht ziehen. Sie können beispielsweise Benutzerinformationen direkt in der Beitragstabelle speichern und so die Notwendigkeit einer Verknüpfung reduzieren.

3. Verwenden Sie Broadcast-Joins für kleine Tabellen: Wenn eine der Tabellen klein genug ist, übertragen Sie sie an alle Knoten, um Scatter-Gather-Abfragen zu vermeiden.

4. Daten vorab zusammenführen: Erwägen Sie bei häufig aufgerufenen Daten die Vorabzusammenführung und Speicherung der Ergebnisse in einer materialisierten Ansicht oder einem Cache.

5. Nutzen Sie verteilte Abfrage-Engines: Verwenden Sie für komplexe analytische Abfragen Systeme wie Presto oder BigQuery, die verteilte Verknüpfungen und Optimierungen automatisch verarbeiten.

---

6. Best Practices für die Cursor-basierte Paginierung mit Sharded-Daten

In einem verteilten System mit einem solchen Sharding muss die cursorbasierte Paginierung sorgfältig gehandhabt werden, insbesondere weil die Daten über mehrere Shards verteilt sind. Der Schlüssel ist:

1. Abfragen aufteilen: Fragen Sie jeden Shard unabhängig nach relevanten Daten ab.
2. Behandeln Sie die Paginierung in Blöcken: Entscheiden Sie, wie in den Shard-Daten paginiert werden soll (entweder bei Beiträgen oder Benutzern), und sammeln Sie relevante Ergebnisse.
3. Auf Anwendungsebene verbinden: Ergebnisse von jedem Shard abrufen, die Daten im Speicher zusammenführen und dann die Cursorlogik für die nächste Seite anwenden.

Lassen Sie uns durchgehen, wie wir dies mit Node.js und Express implementieren können, wobei wir berücksichtigen, dass sich die Daten auf verschiedenen Shards befinden und Post-Fetch-Joins auf Anwendungsebene erfordern.

Umgang mit Paginierung und Verknüpfungen mit Sharded-Tabellen

Nehmen wir an, wir haben:

• Beiträge Tabelle, aufgeteilt nach Benutzer-ID.
• Tabelle Benutzer, aufgeteilt nach Benutzer-ID.

Wir möchten paginierte Beiträge für einen bestimmten Benutzer abrufen, aber da sich Benutzer und Beiträge auf unterschiedlichen Shards befinden, müssen wir die Abfrage aufteilen, die Paginierung durchführen und dann die Verknüpfung auf Anwendungsebene durchführen.

Ansatz:

1. Fragen Sie die relevanten Shards ab:

   • Zuerst müssen Sie die Beitragstabelle über die Shards hinweg abfragen, um die Beiträge abzurufen.
   • Nachdem Sie die relevanten Beiträge abgerufen haben, verwenden Sie die user_id aus den Beiträgen, um die Benutzertabelle abzufragen (wiederum über Shards hinweg).

2. Paginierungsstrategie:

   • Paginierung bei Beiträgen: Sie können „created_at“, „post_id“ oder ein anderes eindeutiges Feld verwenden, um die Beitragstabelle zu paginieren.
   • Paginierung nach Benutzern: Möglicherweise müssen Sie Benutzerdaten separat abrufen oder die Benutzer-ID als Cursor verwenden, um durch Benutzer zu paginieren.

3. Beitritt auf Anwendungsebene:

   • Nachdem Sie Daten aus den relevanten Shards (sowohl für Beiträge als auch für Benutzer) abgerufen haben, verknüpfen Sie diese auf Anwendungsebene.

4. Umgang mit dem Cursor:

   • Verwenden Sie nach dem Abrufen der ersten Seite die zuletzt erstellte_at- oder post_id (aus den Beiträgen) als Cursor für die nächste Abfrage.

Beispielimplementierung

1. Fragen Sie Beiträge über Shards hinweg ab

Hier führen wir Abfragen über verschiedene Beitrags-Shards aus und filtern nach einem Cursor (z. B. erstellt_at oder Beitrags-ID).

2. Fragen Sie Benutzer über Shards hinweg mithilfe von Beitragsdaten ab

Sobald wir die relevante post_id und user_id aus der ersten Abfrage haben, werden wir Benutzerdaten von den relevanten Shards abrufen.

   const express = require('express');
   const { Pool } = require('pg');

   const poolShard1 = new Pool({ connectionString: 'postgresql://localhost/shard1' });
   const poolShard2 = new Pool({ connectionString: 'postgresql://localhost/shard2' });

   const app = express();
   app.use(express.json());

   const getShardPool = (userId) => (userId % 2 === 0 ? poolShard1 : poolShard2);

   app.post('/user', async (req, res) => {
       const { userId, data } = req.body;
       const pool = getShardPool(userId);
       try {
           await pool.query('INSERT INTO user_data (user_id, data) VALUES (, )', [userId, data]);
           res.status(200).send('User added successfully');
       } catch (err) {
           console.error(err);
           res.status(500).send('Error inserting user');
       }
   });

   app.get('/user/:userId', async (req, res) => {
       const userId = parseInt(req.params.userId, 10);
       const pool = getShardPool(userId);
       try {
           const result = await pool.query('SELECT * FROM user_data WHERE user_id = ', [userId]);
           res.status(200).json(result.rows);
       } catch (err) {
           console.error(err);
           res.status(500).send('Error retrieving user');
       }
   });

   app.listen(3000, () => console.log('Server running on port 3000'));

Wichtige Details:

1. Paginierung in Beiträgen: Der Cursor basiert auf dem Feld „created_at“ oder einem anderen eindeutigen Feld in Beiträgen, das zum Paginieren durch Ergebnisse verwendet wird.
2. Shards unabhängig abfragen: Da Beiträge und Benutzer auf verschiedenen Schlüsseln geshardt sind, fragen wir jeden Shard unabhängig ab und sammeln Daten von allen Shards, bevor wir die Verknüpfung auf Anwendungsebene durchführen.
3. Cursor-Handhabung: Nach dem Abrufen der Ergebnisse verwenden wir die zuletzt erstellte_at (oder post_id) aus den Beiträgen, um den Cursor für die nächste Seite zu generieren.
4. Auf Anwendungsebene beitreten: Nachdem wir Daten von den relevanten Shards abgerufen haben, verbinden wir die Beiträge mit Benutzerdaten basierend auf der Benutzer-ID im Speicher.

---

Abschluss

Die Verwaltung fragmentierter Daten in verteilten Systemen stellt einzigartige Herausforderungen dar, insbesondere wenn es um die Durchführung effizienter Verknüpfungen geht. Das Verständnis von Techniken wie Broadcast-Joins, Scatter-Gather-Joins und die Nutzung von verteilten Abfrage-Engines kann die Abfrageleistung erheblich verbessern. Darüber hinaus ist es bei Abfragen auf Anwendungsebene wichtig, Shard-Key-Ausrichtung, Denormalisierung und optimierte Abfragestrategien zu berücksichtigen. Durch die Befolgung dieser Best Practices und den Einsatz der richtigen Tools können Entwickler sicherstellen, dass ihre Anwendungen Sharded-Daten effektiv verarbeiten und die Leistung im großen Maßstab beibehalten.

Das obige ist der detaillierte Inhalt vonUmgang mit Sharded-Daten in verteilten Systemen: Ein tiefer Einblick in Joins, Broadcasts und Abfrageoptimierung. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!