处理分布式系统中的分片数据：深入探讨连接、广播和查询优化

在现代分布式数据库中，水平扩展数据的需求导致了分片的广泛采用。虽然分片有助于管理跨多个节点的大型数据集，但它也带来了挑战，特别是在执行连接并确保高效的数据检索时。在本文中，我们探讨了应对这些挑战的各种概念和技术，特别关注广播连接、分片键对齐和分布式查询引擎，例如Presto 和 BigQuery。此外，我们还演示了如何使用 Node.js 和 Express.

在实际应用程序中处理这些问题

---

Node.js 与 Express.js 中的分片示例

以下是如何使用 Node.js 和 Express.js 在 PostgreSQL 中实现分片。

PostgreSQL 分片示例

使用 Citus 或使用 Node.js 进行手动逻辑分片：

逻辑分片示例

1. 分片设置表:
   使用分片表（shard1 上的 user_data 和 shard2 上的 user_data）。

2. 创建 Express.js API:
   基于分片键（例如 user_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. 分布式数据库中的分片

分片是跨多个数据库实例或分片水平分区数据的过程，以提高性能、可扩展性和可用性。当单个数据库实例无法处理大量数据或流量时，通常需要分片。

分片策略：

• 基于范围的分片：数据根据键的范围分布在分片上，例如，按 order_date 对订单进行分区。
• 基于哈希的分片：通过分片键（例如 user_id）对数据进行哈希处理，以将数据均匀地分布在各个分片上。
• 基于目录的分片：中央目录跟踪数据驻留在系统中的位置。

但是，当相关表在不同键上分片时，或者当一个表需要跨多个分片与另一个表进行联接时，由于需要 分散-聚集 操作，性能可能会下降。这就是理解广播连接和分片键对齐变得至关重要的地方。

---

2. 分片系统中连接的挑战

当数据驻留在不同的分片中时，在这些分片之间执行连接可能会很复杂。以下是常见挑战的细分：

1. 分片键错位：

在许多系统中，表在不同的键上进行分片。例如：

• users 表可能会按 user_id 进行分片。
• 订单表可能按区域进行分片。

执行连接时（例如，orders.user_id = users.user_id），系统需要从多个分片中获取数据，因为相关记录可能不在同一个分片中。

2. 分散-聚集连接：

在分散-聚集连接中，系统必须：

• 向所有保存相关数据的分片发送请求。
• 跨分片的聚合结果。 这会显着降低性能，尤其是当数据分布在许多分片上时。

3. 广播加入：

广播连接 当被连接的一个表足够小，可以广播到所有分片时，就会发生。在这种情况下：

• 小表（例如用户）在较大的分片表（例如订单）所在的所有节点上复制。
• 每个节点都可以将其本地数据与广播数据连接起来，从而避免跨分片通信。

---

3. 使用分布式查询引擎进行分片数据

分布式查询引擎，例如 Presto 和 BigQuery 旨在跨分布式系统高效处理分片数据和联接查询。

急板/Trino：

Presto 是一个分布式 SQL 查询引擎，旨在跨异构数据源（例如关系数据库、NoSQL 数据库、数据湖）查询大型数据集。 Presto 跨分布式数据源执行联接，并可以通过最小化节点之间的数据移动来优化查询。

示例用例：使用 Presto 连接分片数据

在 orders 按区域分片且 users 按 user_id 分片的场景中，Presto 可以使用其分布式执行模型跨不同分片执行连接。

查询:

   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. 使用分散-聚集获取相关用户记录。
2. 跨节点连接数据。

Google BigQuery：

BigQuery 是一个完全托管的无服务器数据仓库，擅长运行大规模分析查询。虽然 BigQuery 抽象了分片的细节，但它会自动在多个节点之间分区和分布数据以优化查询。它可以轻松处理大型数据集，对于数据按时间或其他维度分区的分析查询特别有效。

示例用例：在 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 自动处理分区和分配，最大限度地减少手动分片的需要。

---

4. 处理 Node.js 应用程序中的分片键错位

在 Node.js 应用程序中处理分片数据时，经常会出现 分片键未对齐 以及需要 分散-聚集 连接等问题。以下是如何使用 Node.js 和 Express 应对这些挑战。

在 Node.js 中处理广播连接

如果连接需要在所有分片上广播一个小表（例如用户），您可以通过获取一次小表并使用它与分片表中的数据连接来在应用程序层实现连接。

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

在 Node.js 中处理分散-聚集查询

对于涉及分散-聚集连接的查询（例如，当分片键未对齐时），您将需要查询所有分片并在应用程序层聚合结果。

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. 分片数据查询优化的最佳实践

处理分片数据和执行连接时，请考虑以下最佳实践：

1. 对齐分片键：如果可能，请确保相关表使用相同的分片键。这最大限度地减少了跨分片连接的需求并提高了性能。

2. 反规范化：在频繁连接的场景中，请考虑对数据进行反规范化。例如，您可以将用户信息直接存储在 posts 表中，从而减少联接的需要。

3. 对小表使用广播连接：如果其中一个表足够小，则将其广播到所有节点以避免分散-聚集查询。

4. 预连接数据：对于经常访问的数据，考虑预连接并将结果存储在物化视图或缓存中。

5. 利用分布式查询引擎：对于复杂的分析查询，请使用 Presto 或 BigQuery 等自动处理分布式联接和优化的系统。

---

6. 使用分片数据进行基于游标的分页的最佳实践

在具有此类分片的分布式系统中，基于游标的分页需要谨慎处理，特别是因为数据分布在多个分片中。关键是：

1. 拆分查询：独立查询每个分片以获取相关数据。
2. 分页处理：决定如何对分片数据（帖子或用户）进行分页，并收集相关结果。
3. 在应用程序级别加入：从每个分片中获取结果，在内存中加入数据，然后应用下一页的游标逻辑。

让我们来看看如何使用 Node.js 和 Express 来实现这一点，考虑到数据驻留在不同的分片上并且需要在应用程序级别进行提取后连接。

如何处理分页和与分片表的连接

假设我们有：

• 帖子按user_id分片的表。
• users 按 user_id 分片的表。

我们想要检索给定用户的分页帖子，但由于用户和帖子位于不同的分片上，因此我们需要拆分查询、处理分页，然后在应用程序级别执行联接。

方法：

1. 查询相关分片:

   • 首先，您需要跨分片查询 posts 表以获取帖子。
   • 获取相关帖子后，使用帖子中的 user_id 查询用户表（同样是跨分片）。

2. 分页策略:

   • 帖子分页：您可以使用created_at、post_id或其他唯一字段对帖子表进行分页。
   • 用户分页：您可能需要单独获取用户数据或使用 user_id 作为游标来对用户进行分页。

3. 应用程序级连接:

   • 从相关分片（帖子和用户）检索数据后，在应用程序级别加入它们。

4. 处理光标:

   • 获取第一页后，使用最后的created_at或post_id（来自帖子）作为下一个查询的光标。

实施示例

1. 跨分片查询帖子

在这里，我们将跨不同的帖子分片执行查询，并通过游标（例如，created_at 或 post_id）进行过滤。

2.使用Post数据跨分片查询用户

一旦我们从第一个查询中获得了相关的 post_id 和 user_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. 帖子分页：光标基于created_at字段或帖子中的另一个唯一字段，用于对结果进行分页。
2. 独立查询分片：由于帖子和用户在不同的键上分片，因此我们独立查询每个分片，在应用程序级别执行联接之前从所有分片收集数据。
3. 光标处理：检索结果后，我们使用帖子中的last created_at（或post_id）来生成下一页的光标。
4. 在应用程序级别加入：从相关分片获取数据后，我们根据内存中的 user_id 将帖子与用户数据加入。

---

结论

管理分布式系统中的分片数据提出了独特的挑战，特别是在执行高效连接时。了解广播连接、分散-聚集连接等技术并利用分布式查询引擎可以显着提高查询性能。此外，在应用程序级查询中，必须考虑分片键对齐、非规范化和优化的查询策略。通过遵循这些最佳实践并利用正确的工具，开发人员可以确保他们的应用程序有效处理分片数据并大规模维持性能。

以上是处理分布式系统中的分片数据：深入探讨连接、广播和查询优化的详细内容。更多信息请关注PHP中文网其他相关文章！