数据流简介

那么让我们从什么是数据开始？，它是已被翻译成对移动或处理有效的形式的信息。相对于今天的计算机和传输介质，数据是转换成二进制数字形式的信息。
据统计，全球每天创建的文本数据量超过 180 亿条。因此，数据一直在全球各地的计算机之间创建和传输。
数据可以通过多种方式传输，要理解数据流传输，我们需要了解传统方式并了解为什么我们需要流式传输。

传输文件的一种常见方法是所有数据包必须到达目的地进行重组，然后到达目的地，例如当你发送图像并且你需要该图像的所有位来显示它因此必须交付它第一的。

但是，如果你正在等待视频加载，你是否正在等待完整的视频下载和播放？不是我们所有人，这就是为什么我们拥有 Youtube、Netflix 等流媒体服务，你可以立即开始播放其中的任何一个，这就是流媒体创意发挥作用的地方。

流的独特之处在于，不是程序像传统方式那样一次将文件全部读入内存，而是逐块读取数据块，处理其内容而不将其全部保存在内存中。
无需等待 100 MB 的视频加载，我们可以将其逐块加载并加载每个 1MB，因为它们从流中被消耗并立即开始显示。

数据库的数据流是什么意思？

由于我们清楚地理解了数据流的含义，以及它将如何在数据库世界中帮助我们，这是我们的下一个主题。

假设我们有一个来自 SQL 的表，其中包含一些数据（大约 1000 条），并且你创建一个 select 语句来检索它们以进行一些 O(N) 计算，它足够快，你不会对报告感到任何延迟。

但是，假设我们有 1B 行……你的 select 语句需要足够的时间，而你的 O ( N ) 计算需要额外的时间，因为它必须等待检索行，然后开始遍历所有行来执行计算。

现在有趣的部分，我们如何改进呢？是的，你没看错，让我们看看这个 select 语句的流将如何帮助我们。
与其等待 10 亿行，不如逐个获取一行，这样当数据库获取一行时，我们会立即对其进行一些计算，因为它们是该数据的一大块，我们逐块处理，然后我们在接收下一个块之前将其发送回用户。

到此提取结束时，你将有一个 1B 的计算行被发送回用户并且用户根本没有等待，我们只优化了 O ( N ) 计算，这将极大地改善你更复杂的计算.

数据库流将减少用户等待时间并优化你对大量数据的计算。

注意：这只是对我们文章的解释，因为我们在下一节中会更深入，数据库流对创建管道和数据湖的数据科学也有巨大的影响，但这是另一篇文章

可以用 MySQL 的数据流解决的案例

流式传输你的数据库数据可以有许多应用程序。我通常将它用于需要对数十亿行进行计算并且必须以快速方式完成的业务所需的计算和报告。

也许你正在为大数据创建 Excel 文件，或者为大型报告创建 PDF，然后在获取它们时将其插入其中会更快。

你可能想要一个视频播放器并将二进制数据存储在数据库中，然后你可以将其流式传输回给用户。你可能有一个从数据库中获取图像的图库，你可以逐个显示图像或更多应用程序。

如果你是数据科学家并尝试自己创建管道以在两个数据库之间迁移数据，那么你可以每天流式传输它们以保持同步，或者如果你正在寻找数据湖来更改数据，那么你可以修改块在流式传输数据时按块。

简而言之，在任何情况下，你都可以将当前结构改进得更快。

Knex.js 简介（Node.js 的 NPM 包）

在编码我们的流媒体之前，我们将使用 Knex.js 库作为我们的查询构建器。

Knex.js（你知道他们在前面也发音为“K”吗？哈哈）是一个“包含电池”的 SQL 查询构建器，适用于 PostgreSQL、CockroachDB、MSSQL、MySQL、MariaDB、SQLite3、Better-SQLite3、Oracle 和 Amazon Redshift设计灵活、便携且有趣。

他们提供了一种在 Node.js 中使用 SQL 的漂亮方式，你可以参考他们的官方文档来了解更多关于这个神奇产品的信息。

你可以在此处查看文档：Knex.js 文档
，你可以查看流式文档：Knex.js 流文档

Knex.js 的一项功能是 Streams for MySQL，你可以使用 javascript 以非常简单的方式对查询进行流式处理和修改。

然而，Knex 使用原始 Mysql npm 为 node.js 提供的流，该流由 Felix Geisendörfer 和他的出色团队制作，这使得 Knex 更容易将他们的流使用到他们的库中。我们将在后面的部分讨论 Mysql 包是如何实现这一点的。

Node.js 中的 Streaming 简介

最后值得一提的是 node.js Streaming 模块，对于我们将在下一节中进行的实现，node.js 流在其中也有它的作用以及 MySQL 提供的功能，所以让我们简要解释一下什么是node.js 流？

Stream 模块是 Node.js 中默认提供的原生模块。Stream 是 EventEmitter 类的一个实例，它在 Node.js 中异步处理事件。由于它们的超类，流本质上是基于事件的。

Node.js 中有 4 种类型的流：

Writable：用于顺序写入数据
Readable：用于顺序读取数据
Duplex：用于顺序读取和写入数据
Transform：在写入或读取时可以修改数据。以压缩为例，使用这样的流，你可以写入压缩数据并读取解压缩数据。

这是关于 node.js 流模块的简要介绍，有关更多信息，你可以在Node.js 官方文档中阅读有关流的所有信息。

使用 Node.js 实现 MySQL 的数据流

在本节中，我们将使用 Knex.js 包对我们的流媒体进行编码，所以让我们马上开始吧。

作为旁注，Node.js 和 Knex.js 包的基本使用知识是必需的，因为我将只关注整个编码过程中的流。

首先，我将创建一个名为“stream.js”的文件，其中包含一个名为“sample”的异步函数，这将是本文中的示例。

const database = require('./database/connection');
 async function sample(){
    console.log("Hi , This is a sample function");
 }
 sample();

我们在顶部有 knex mysql 连接，我可以使用“node stream.js”运行这个文件

然后，我将从 SQL 创建一个表，以便我们可以在.. 上编写查询。我将为它快速编写迁移并向其中添加一些测试数据。

CREATE TABLE `sample` (
  `id` int unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(255) NOT NULL DEFAULT '',
  PRIMARY KEY (`id`)
)

我确实在这个数据库中添加了一些虚拟数据，大约 3000 条记录

SELECT COUNT(*) FROM sample; //returns 3000

现在我将使用 Knex.js Stream 函数来修改每一个，因为它们被提取

const database = require("./database/connection");
async function sample() {
  console.log("Started At :", new Date().toISOString());
  await database("sample")
    .select()
    .stream((stream) => {
      stream.on("data", (row) => {
        console.log(row.name, `At : ${new Date().toISOString()}`);
      });
    });
    console.log("Ended At :", new Date().toISOString());
}
sample();

运行此代码将打印从数据库中获取的行，并带有确切的开始日期和结束日期，你可以在它们之间做出区分。

我们所做的，我们只是为这个表写了一个 select *，然后我们使用了 knex.js 提供的 .stream 函数，然后我们正在监听“data”，它指示每一行何时到达，还有其他事件，如 on Error处理流中发生的错误。