# @electric-sql/d2ts 中文文档

规范地址：https://npmcn.com/package/%40electric-sql/d2ts/

版本：0\.1\.8
npm 元数据更新时间：2026-01-30T21:50:01.116Z
中文译文更新时间：2026-11-13T11:07:39.856Z
软件包许可：Apache-2\.0

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40electric-sql/d2ts
原文对照：https://npmcn.com/package/%40electric-sql/d2ts/?lang=original
项目主页：https://github.com/electric-sql/d2ts
源码仓库：https://github.com/electric-sql/d2ts

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：1.1 MB，234 个文件
- 模块格式：ES Module
- TypeScript 类型：包内自带
- 运行依赖：3 个
- peer 依赖：better-sqlite3 ^11\.7\.0，@electric-sql/client \>=1\.0\.0-beta\.4
- 首次发布：2025-02-26T22:00:16.559Z
- 最新版本发布：2025-06-13T19:59:32.686Z
- 近 30 天下载量：11487（2026-10-14 至 2026-11-12，npm 官方下载统计）
- 已发布版本数：9（近 12 个月 0 个）

## 中文简介

D2TS是 Difficial Data flow 的一种类型脚本执行.

## 安装

```sh
npm install @electric-sql/d2ts
```

也可以使用：`pnpm add @electric-sql/d2ts`、`yarn add @electric-sql/d2ts`、`bun add @electric-sql/d2ts`

## README

[![ElectricSQL logo](https://raw.githubusercontent.com/electric-sql/meta/main/identity/ElectricSQL-logo-black.svg)](https://electric-sql.com/)

[![](https://github.com/electric-sql/d2ts/actions/workflows/ci.yml/badge.svg)](https://github.com/electric-sql/d2ts/actions) [![License - Apache 2.0](https://img.shields.io/badge/license-Apache_2.0-green)](https://github.com/electric-sql/d2ts/blob/main/LICENSE) [![Status - Alpha](https://img.shields.io/badge/status-alpha-orange)](https://github.com/electric-sql/d2ts/%20%20ext/milestones) [![Chat - Discord](https://img.shields.io/discord/933657521581858818?color=5969EA&label=discord)](https://discord.electric-sql.com/) [![](https://img.shields.io/twitter/follow/ElectricSQL.svg?style=social&label=Follow%20@ElectricSQL)](https://x.com/ElectricSQL)

# D2TS - 类型脚本中的差异数据流

D2TS 是类型脚本执行 [差异数据流量](https://github.com/MaterializeInc/differential) - 一个强大的数据并行编程框架,能够进行递增计算而不是不断变化的输入数据.

您可以使用 D2TS 来构建可以递增执行的数据管道,这意味着您可以在输入时处理数据,并且只能重算已经改变的部分. 这可能与重新绘制数据一样简单,或者像完全结合两个数据源那样复杂,因为一个数据源是一个计算的总和.

D2TS 可与 [电子SQL](https://electric-sql.com/) 将数据管道建在 [形状结构](https://electric-sql.com/docs/api/clients/typescript#shapestream) 可逐步执行.

一条D2TS管线也是完全型式安全的,推断出管线每一步的型号,支持自动完成您的 IDE.

## 目录

-   [关键特性](https://npmcn.com/package/%40electric-sql/d2ts/#readme-key-features)
-   [快速启动](https://npmcn.com/package/%40electric-sql/d2ts/#readme-quick-start)
-   [实例](https://npmcn.com/package/%40electric-sql/d2ts/#readme-examples)
-   [API 密码](https://npmcn.com/package/%40electric-sql/d2ts/#readme-api)
    -   [D2 图解](https://npmcn.com/package/%40electric-sql/d2ts/#readme-d2-graph-construction)
    -   [输入流](https://npmcn.com/package/%40electric-sql/d2ts/#readme-input-streams)
    -   [版本和边框](https://npmcn.com/package/%40electric-sql/d2ts/#readme-versions-and-frontiers)
    -   [以更改方式多选项](https://npmcn.com/package/%40electric-sql/d2ts/#readme-multiset-as-a-changeset)
    -   [运算符](https://npmcn.com/package/%40electric-sql/d2ts/#readme-operators)
    -   [使用 SQLite 后端](https://npmcn.com/package/%40electric-sql/d2ts/#readme-using-sqlite-backend)
-   [执行细节](https://npmcn.com/package/%40electric-sql/d2ts/#readme-implementation-details)
-   [参考资料](https://npmcn.com/package/%40electric-sql/d2ts/#readme-references)

## 关键特性

-   **递增处理**: 高效处理输入数据的更改而无需重算所有数据
-   **丰富的运算符**: 以管道API支持共同业务:
    -   [`buffer`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-buffer): 完成后缓冲并放出版本
    -   [`concat`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-concat): 连通两条溪流
    -   [`consolidate`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-consolidate): 每个版本合并流中的元素
    -   [`count`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-count): 按键计算元素
    -   [`distinct`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-distinct): 删除重复
    -   [`filter`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-filter): 基于上游的过滤元素
    -   [`filterBy`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-filterby): 用另一流的密钥过滤密钥流的元素
    -   [`iterate`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-iterate): 进行迭代计算
    -   [`join`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-join): 加入两个溪流
    -   [`keyBy`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-keyBy): 由属性键入流
    -   [`map`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-map): 流中的变换元素
    -   [`reduce`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-reduce): 按密钥分列的总值
    -   [`rekey`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-rekey): 更改密钥流的密钥
    -   [`unkey`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-unkey): 从密钥流中删除密钥
    -   [`output`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-output): 输出流中的消息
    -   [`pipe`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-pipe): 建立运营商管道,使运营商组合得以再利用
    -   [`topK`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-topk): 将结果限制在以比较器为基础的K大元素
    -   [`topKWithIndex`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-topkwithindex):类似于上K,但包括位置指数
    -   [`topKWithFractionalIndex`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-topkwithfractionalindex): 如上KWithIndex,但分数指数稳定
    -   [`groupBy`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-groupby): 按密钥分组数据并应用多个聚合函数
        -   [`sum`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-sum): 每一组中的总和值
        -   [`count`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-count): 计算每组项目
        -   [`avg`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-avg): 计算每组平均值
        -   [`min`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-min): 在每个组中查找最小值
        -   [`max`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-max): 在每个组中查找最大值
        -   [`median`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-median): 计算每组的中值
        -   [`mode`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-mode): 在每个组中查找最常见的值
    -   [`orderBy`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-orderby): 按取值函数排列元素
    -   [`orderByWithIndex`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-orderbywithindex): 排列要素并包括位置指数
    -   [`orderByWithFractionalIndex`](https://npmcn.com/package/%40electric-sql/d2ts/#readme-orderbywithfractionalindex): 分数指数稳定的元素顺序
-   **SQLite 整合**: 持续操作状态的可选 SQLite 后端,允许更大的数据集和可回收管道
-   **类型安全**: 完全类型Script 类型安全和通过管道API推论

## 快速启动

### 安装

```bash
npm install @electric-sql/d2ts
```

### 基本用途

以下是一个简单的例子,说明核心概念:

```typescript
import { D2, map, filter, debug, MultiSet, v } from '@electric-sql/d2ts'

// Create a new D2 graph with initial frontier
// The initial frontier is the lower bound of the version of the data that may
// come in future.
const graph = new D2({ initialFrontier: 0 })

// Create an input stream
// We can specify the type of the input stream, here we are using number.
const input = graph.newInput<number>()

// Build a simple pipeline that:
// 1. Takes numbers as input
// 2. Adds 5 to each number
// 3. Filters to keep only even numbers
// Pipelines can have multiple inputs and outputs.
const output = input.pipe(
  map((x) => x + 5),
  filter((x) => x % 2 === 0),
  debug('output'),
)

// Finalize the pipeline, after this point we can no longer add operators or
// inputs
graph.finalize()

// Send some data
// Data is sent as a MultiSet, which is a map of values to their multiplicity
// Here we are sending 3 numbers (1-3), each with a multiplicity of 1
// When you send data, you set the version number, here we are using 0
// The key thing to understand is that the MultiSet represents a *change* to
// the data, not the data itself. "Inserts" and "Deletes" are represented as
// an element with a multiplicity of 1 or -1 respectively.
input.sendData(
  0, // The version of the data
  new MultiSet([
    [1, 1],
    [2, 1],
    [3, 1],
  ]),
)

// Set the frontier to version 1
// The "frontier" is the lower bound of the version of the data that may come in future.
// By sending a frontier, you are indicating that you are done sending data for any version less than the frontier and therefor D2TS operators that require them can process that data and output the results.
input.sendFrontier(1)

// Process the data
graph.run()

// Output will show:
// 6 (from 1 + 5)
// 8 (from 3 + 5)
```

### 使用电源SQL

D2TS 可与 [电子SQL](https://electric-sql.com/) 将数据管道建在 [形状结构](https://electric-sql.com/docs/api/clients/typescript#shapestream) 可逐步执行.

> \[注\] 电源支持尚未合并为主, 您可以跟踪进度 [在本 PR 中](https://github.com/electric-sql/d2ts/pull/11)。 。 。 .

以下是使用ElectricSQL的D2TS的例子:

```typescript
import { D2, map, filter, output } from '@electric-sql/d2ts'
import { electricStreamToD2Input } from '@electric-sql/d2ts/electric'
import { ShapeStream } from '@electric-sql/client'

// Create D2 graph
const graph = new D2({ initialFrontier: 0 })

// Create D2 input
const input = graph.newInput<any>()

// Configure the pipeline
input.pipe(
  map(([key, data]) => data.value),
  filter((value) => value > 10),
  // ... any other processing / joining
  output((msg) => doSomething(msg)),
)

// Finalize graph
graph.finalize()

// Create Electric stream (example)
const electricStream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
    replica: 'full', // <-- IMPORTANT!
  },
})

// Connect Electric stream to D2 input
electricStreamToD2Input(electricStream, input)
```

## 实例

本文中有一些实例 [./实例](https://npmcn.com/package/%40electric-sql/d2ts/examples) 目录,涵盖:

-   [基本用途](https://npmcn.com/package/%40electric-sql/d2ts/examples/basic.ts) (图和过滤器)
-   ["处理的Fruit"](https://npmcn.com/package/%40electric-sql/d2ts/examples/fruit-processed.ts) (减少和合并)
-   [在两个溪流之间组合](https://npmcn.com/package/%40electric-sql/d2ts/examples/join.ts)
-   [迭代计算](https://npmcn.com/package/%40electric-sql/d2ts/examples/iterate.ts)
-   [使用加入来模拟“ 包含”](https://npmcn.com/package/%40electric-sql/d2ts/examples/includes.ts)

## API 密码

### D2 图解

```typescript
const graph = new D2({ initialFrontier: 0 })
```

这个 `D2` 构造器使用可选 `options` 具有下列属性的对象:

-   `initialFrontier`: 图中的初始边框,默认为 `0`

一个D2图的例子用于构建数据流图,主要方法如下:

-   `newInput<T>(): IStreamBuilder<T>`: 创建新输入流
-   `finalize(): void`: 最后确定图表, 在此点之后, 不再添加运算符或输入
-   `run(): void`: 处理所有待处理的数据流图版本

### 输入流

使用 `newInput<T>()` 方法,并具有下列方法:

-   `sendData(version: Version | number | number[], data: MultiSet<T>): void`: 向输入流发送数据
-   `sendFrontier(version: Antichain | Version | number | number[]): void`: 向输入流发送边框

### 版本和边框

版本用于代表数据版本,是一团整数. 对于大多数使用的案例,您只需要提供单个整数版本,所有使用一个版本的APIS都会使用单个整数工作. 更先进的使用案例可能要求使用纬度来跟踪多维版本.

边框用于代表未来可能来的数据版本的下限,是版本的反链. 在大多数情况下,你只能使用一个单一的整数版本来代表边框.

#### 版本

有一个 `Version` 类代表一个版本,创建一个版本的首选方式是使用 `v` 帮助器的功能可以确保您为同一版本重新使用同一对象,从而使平等检查和比较更加有效:

```typescript
const version = v(1)
```

多层面版本也得到支持,并使用 `v` 帮助函数 :

```typescript
const version = v([1, 2])
```

在多数情况下,您只需要使用单个整数版本来代表可直接传递到 `sendData` 财务报告和审定财务报表 `sendFrontier` 方法:

```typescript
input.sendData(1, new MultiSet([[1, 1]]))
```

#### 反链(边境)

一个 `Antichain` 是一组不相连的版本,用来代表数据的前沿。 使用 `Antichain` 构造器 :

```typescript
const frontier = new Antichain([v(1), v([2])])
```

在大多数情况下,您只需要使用单个整数版本来代表边框,就可以直接传递给 `sendFrontier` 方法 :

```typescript
input.sendFrontier(1)
```

### 以更改方式多选项

A 类 `MultiSet` 是一幅价值图,以示其多重性。 它用来代表对集合的修改.

多集程序是通过传递一系列 `[value, multiplicity]` 双人 在这里,我们正在创建一个多集,其中值为1,2和3,每集为一:

```typescript
const multiSet = new MultiSet([
  [1, 1],
  [2, 1],
  [3, 1],
])
```

MultiSets可用于表示任何对象:

```typescript
// Here we have a MultiSet of new "comments" with the interface `Comment`
const multiSet = new MultiSet<Comment>([
  [{ id: '1', text: 'Hello, world!', userId: '321' }, 1],
  [{ id: '2', text: 'Hello, world!', userId: '123' }, 1],
])
```

D2TS的一个重要原则是"键"多Sets,其中: `value` 是两倒数 `[key, value]`。 。 。 .

```typescript
// Here we have a MultiSet of new "comments" but we have keyed them by the
// `userId`
const multiSet = new MultiSet<[string, Comment]>([
  [['321', { id: '1', text: 'Hello, world!', userId: '321' }], 1],
  [['123', { id: '2', text: 'Hello, world!', userId: '123' }], 1],
])
```

插入和删除分别作为1或1的多元素表示.

```typescript
// Here we are inserting a one new comment and deleting one comment
const multiSet = new MultiSet<[string, Comment]>([
  [['321', { id: '1', text: 'Hello, world!', userId: '321' }], 1],
  [['123', { id: '2', text: 'Hello, world!', userId: '123' }], -1],
])
```

### 运算符

#### 缓冲

`buffer()`

缓冲流的元素,当缓冲完成时会释放出一个版本.

```typescript
const output = input.pipe(buffer())
```

#### 弯曲

`concat(other: IStreamBuilder<T>)`

凝聚出两个输入流 - 输出流会包含两个流的元素.

```typescript
const output = input.pipe(concat(other))
```

#### 合并

`consolidate()`

在每个版本整合流中的元素,基本上确保输出流最晚为已知 _完整_ 版本.

```typescript
const output = input.pipe(consolidate())
```

#### 计数

`count()`

按键计算流中的元素数

```typescript
const output = input.pipe(
  map((data) => [data.somethingToKeyOn, data]),
  count(),
)
```

#### 调试

`debug(name: string)`

将流信息记录到控制台,名称用于识别日志中的流.

```typescript
const output = input.pipe(debug('output'))
```

#### 独立

`distinct()`

从流中按密钥删除重复值

```typescript
const output = input.pipe(distinct())
```

#### 过滤器

`filter(predicate: (data: T) => boolean)`

基于上游过滤流

```typescript
const output = input.pipe(filter((x) => x % 2 === 0))
```

#### 过滤By

`filterBy(other: IStreamBuilder<KeyValue<K, unknown>>)`

过滤按键流的元素,按另一个流的密钥进行。 这使得您可以在您有多个相关产出的地方建立管道,例如项目的问题流和评论.

```typescript
// Create two input streams
const issues = graph.newInput<KeyValue<string, Issue>>()
const comments = graph.newInput<KeyValue<string, Comment>>()

// Filter comments to only include those related to specific issues
const commentsForIssues = comments.pipe(
  keyBy((comment) => comment.issue_id),
  filterBy(issues),
  rekey((comment) => comment.id),
  output((message) => {
    // Process only comments that have matching issues
    console.log('Comment for tracked issue:', message)
  }),
)
```

在这个例子中,只有带有问题流中存在的密钥的评论才会被收录到产出中.

#### 斜率

`iterate<T>(f: (stream: IStreamBuilder<T>) => IStreamBuilder<T>)`

通过创建反馈回路在流上进行迭代计算. 这使得您可以反复处理数据,直到其达到固定点或满足特定条件.

这个 `iterate` 运算符使用定义迭代步骤的函数。 在此函数内部,您可以将任何一系列的变换应用到流中,结果会被反馈回回循环中进行进一步的迭代.

```typescript
// This example repeatedly doubles numbers and includes previous values,
// filtering out any values > 50
const output = input.pipe(
  iterate((stream) =>
    stream.pipe(
      map((x) => x * 2), // Double each value
      concat(stream), // Include original values
      filter((x) => x <= 50), // Keep only values <= 50
      map((x) => [x, []]), // Convert to keyable format
      distinct(), // Remove duplicates
      map((x) => x[0]), // Convert back to simple values
      consolidate(), // Ensure consistent version updates
    ),
  ),
  debug('results'),
)
```

举例来说:

1.  这个 `iterate` 函数在输入流上创建反馈循环
2.  每个值都加倍,然后与所有以前的值合并
3.  过滤出大于50的值
4.  剩余值在下次重复之前被复制并合并

重复将持续到没有产生出新的值(影响一个固定点)或直到边框超越了重复范围.

常用案例 `iterate` 运算符包括:

-   图表算法中的计算机过渡关闭
-   在趋同之前宣传价值
-   执行固定点算法
-   以限定结果模拟递归过程

这个强大的操作员能够进行复杂的递归计算,同时保持差分数据流的递增性.

#### 加入时

`join(other: IStreamBuilder<T>, joinType?: JoinType)`

基于匹配的密钥加入两个密钥流 。 这个 `joinType` 参数控制加盟行为:

-   `'inner'` (默认):只返回两个流中有匹配密钥的记录
-   `'left'`: 返回从左侧流出的所有记录, 加上右侧的匹配记录( 非匹配的无效)
-   `'right'`: 返回从右侧流出的所有记录, 加上从左侧取出匹配的记录( 非匹配的无效)
-   `'full'`: 返回来自两条流的所有记录, 双方无匹配记录为空

```typescript
const input = graph.newInput<[key: string, value: number]>()
const other = graph.newInput<[key: string, value: string]>()

// Inner join - only matching keys
const innerJoin = input.pipe(join(other, 'inner'))

// Left join - all records from input, matching from other
const leftJoin = input.pipe(join(other, 'left'))

// Right join - all records from other, matching from input
const rightJoin = input.pipe(join(other, 'right'))

// Full join - all records from both streams
const fullJoin = input.pipe(join(other, 'full'))
```

此外,还有加盟运营商的型号安全变体(`leftJoin`, (中文(简体) ), `rightJoin`, (中文(简体) ), `fullJoin`, (中文(简体) ), `antiJoin`),对不同的关联类型有适当的可撤销类型:

```typescript
// The two streams are initially keyed by the userId and commentId respectively
const comments = graph.newInput<[commentId: string, comment: Comment]>()
const users = graph.newInput<[userId: string, user: User]>()

// Map the comments to be "keyed" by the user id
const commentsByUser = comments.pipe(
  map(([commentId, comment]) => [comment.userId, comment] as [string, Comment]),
)

// Left join - keeps all comments, even those without matching users
const output = commentsByUser.pipe(
  leftJoin(users),
  map(([userId, [comment, user]]) => {
    // user can be null in a left join if there's no matching user
    return [
      comment.id,
      {
        ...comment,
        userName: user?.name ?? 'Unknown User',
      },
    ]
  }),
)
```

当使用 SQLite 持久性时, 您可以作为附加参数提供数据库 :

```typescript
// Using SQLite persistence
const db = new BetterSQLite3Wrapper(sqlite)
const persistedJoin = input.pipe(join(other, 'inner', db))
```

#### 地图

`map<U>(f: (data: T) => U)`

使用函数来转换流中的元素

```typescript
const output = input.pipe(map((x) => x + 5))
```

#### 产出

`output(messageHandler: (message: Message<T>) => void)`

输出流中的消息

```typescript
input.pipe(
  output((message) => {
    if (message.type === MessageType.DATA) {
      console.log('Data message', message.data)
    } else if (message.type === MessageType.FRONTIER) {
      console.log('Frontier message', message.data)
    }
  }),
)
```

信息是一个 `Message<T>` 对象,其结构:

```typescript
type Message<T> =
  | {
      type: typeof MessageType.DATA
      data: DataMessage<T>
    }
  | {
      type: typeof MessageType.FRONTIER
      data: FrontierMessage
    }
```

数据电文代表输出数据的变化,具有下列数据有效载荷:

```typescript
type DataMessage<T> = {
  version: Version
  collection: MultiSet<T>
}
```

前沿电文代表了一个新的前沿,具有以下数据有效载荷:

```typescript
type FrontierMessage = Version | Antichain
```

#### 烟斗

`pipe(operator: (stream: IStreamBuilder<T>) => IStreamBuilder<T>)`

管道流通过一系列运算符

```typescript
// You can specify the input and output types for the pipeline.
// Here we are specifying the input type as number and the output type as
// string.
const composedPipeline = pipe<number, string>(
  map((x) => x + 5),
  filter((x) => x % 2 === 0),
  map((x) => x.toString()),
  debug('output'),
)

const output = input.pipe(
  map((x) => x + 1),
  composedPipeline,
)

// Or as a function

const myPipe = (a: number, b: number) =>
  pipe<number, number>(
    map((x) => x + a),
    filter((x) => x % b === 0),
    debug('output'),
  )

const output = input.pipe(myPipe(5, 2))
```

#### 减少

`reduce(f: (values: [T, multiplicity: number][]) => [R, multiplicity: number][])`

在按密钥分组的流上执行一个减少操作 .

职能 `f` 取出一系列值及其相乘性,并返回一系列结果及其相乘性.

```typescript
// Sum a values by key from the input stream
const output = input.pipe(
  map((data) => [data.somethingToKeyOn, data.aValueToSum]),
  reduce((values) => {
    // `values` is an array of [value, multiplicity] pairs for a specific key
    let sum = 0
    for (const [value, multiplicity] of values) {
      sum += value * multiplicity
    }
    return [[sum, 1]]
  }),
  output((message) => {
    if (message.type === MessageType.DATA) {
      // `message.data` is a MultiSet representing the changes to the output
      // data
      // In this example, the output stream will contain the change to the
      // sum of the values for each key.
      console.log(message.data)
    }
  }),
)
```

#### 密钥运算符

D2TS提供一组操作器,用于与密钥流合作,这些流对连接和分组等操作有用.

##### `keyBy`

以每个元素的属性键入流。 这对于为加入或分组操作准备数据很有用.

```typescript
const keyedStream = input.pipe(keyBy((item) => item.id))
```

##### `unkey`

从密钥流中去除密钥,只返回值.

```typescript
const unkeyedStream = keyedStream.pipe(unkey())
```

##### `rekey`

根据值属性将密钥流的密钥更改为新密钥 .

```typescript
const rekeyedStream = keyedStream.pipe(rekey((item) => item.newKey))
```

加入的示例用法 :

```typescript
// Transform comments into [issue_id, comment] pairs for joining
const commentsByIssue = inputComments.pipe(rekey((comment) => comment.issue_id))

// Join comments with issues
const issuesWithComments = issuesForProject.pipe(join(commentsByIssue))
```

#### 组字

`groupBy(keyExtractor: (data: T) => K, aggregates: Record<string, AggregateFunction<T, any>>)`

按密钥分组数据,并对每个组应用多个集合函数. 这是一个强大的操作器,可以结合类似于 SQL 的功能 `GROUP BY` 带有聚合函数.

这个 `keyExtractor` 函数从每个数据项中提取组合键,并 `aggregates` 对象映射集合名称到集合函数.

```typescript
// Group sales data by category and compute multiple aggregates
const salesAnalytics = salesData.pipe(
  groupBy(
    // Key extractor function - what to group by
    (item) => ({ category: item.category }),
    // Aggregate functions to apply to each group
    {
      total: sum((item) => item.amount),
      count: count(),
      average: avg((item) => item.amount),
      min: min((item) => item.amount),
      max: max((item) => item.amount),
    },
  ),
)
```

其结果将是一组带有组合密钥属性和计算总值的对象:

```typescript
// Example output item
{
  category: "Electronics",
  total: 2500,
  count: 5,
  average: 500,
  min: 200,
  max: 800
}
```

##### 总函数

D2TS 提供了几种内置的集合函数,用于: `groupBy`数字 :

###### `sum`

`sum<T>(valueExtractor?: (value: T) => number)`

计算每组数值的总和.

```typescript
// Sum the amount field for each category
groupBy((item) => ({ category: item.category }), {
  total: sum((item) => item.amount),
})
```

###### `count`

`count<T>()`

计算每组项目的数量.

```typescript
// Count items in each category
groupBy((item) => ({ category: item.category }), {
  itemCount: count(),
})
```

###### `avg`

`avg<T>(valueExtractor?: (value: T) => number)`

计算每组数值的平均值.

```typescript
// Calculate average price by category
groupBy((item) => ({ category: item.category }), {
  averagePrice: avg((item) => item.price),
})
```

###### `min`

`min<T>(valueExtractor?: (value: T) => number)`

在每个组中找到最小值.

```typescript
// Find minimum price by category
groupBy((item) => ({ category: item.category }), {
  lowestPrice: min((item) => item.price),
})
```

###### `max`

`max<T>(valueExtractor?: (value: T) => number)`

在每个组中查找最大值.

```typescript
// Find maximum price by category
groupBy((item) => ({ category: item.category }), {
  highestPrice: max((item) => item.price),
})
```

###### `median`

`median<T>(valueExtractor?: (value: T) => number)`

计算每个组的中位数(中值)。 对于数值偶数的组,返回两个中值的平均值.

```typescript
// Calculate median price by category
groupBy((item) => ({ category: item.category }), {
  medianPrice: median((item) => item.price),
})
```

###### `mode`

`mode<T>(valueExtractor?: (value: T) => number)`

查找每个组中最频繁的值。 如果多个值的频率相同,则返回遇到的第一个值.

```typescript
// Find most common price point by category
groupBy((item) => ({ category: item.category }), {
  mostCommonPrice: mode((item) => item.price),
})
```

##### 组合多个聚合

权力 `groupBy` 源自一个单一操作中多个集合的组合:

```typescript
// Comprehensive sales analysis by region and product category
const salesAnalysis = salesData.pipe(
  groupBy(
    // Group by both region and category
    (sale) => ({
      region: sale.region,
      category: sale.productCategory,
    }),
    // Apply multiple aggregates
    {
      totalRevenue: sum((sale) => sale.amount),
      orderCount: count(),
      averageOrderValue: avg((sale) => sale.amount),
      smallestOrder: min((sale) => sale.amount),
      largestOrder: max((sale) => sale.amount),
      medianOrder: median((sale) => sale.amount),
      mostCommonAmount: mode((sale) => sale.amount),
    },
  ),
)
```

#### 顺序

`orderBy(valueExtractor: (value: V) => T, options?: { comparator?: (a: T, b: T) => number, limit?: number, offset?: number })`

以从每个元素中提取的值对按键流中的元素进行排序,可选限制并抵消. 这个操作员命令整个溪流,而不仅仅是在关键组内.

```typescript
// Order all products by price
const orderedProducts = products.pipe(
  map((product) => [product.category, product]), // Key by category
  orderBy((product) => product.price, {
    comparator: (a, b) => a - b, // Sort by price ascending
    limit: 10, // Only keep top 10
  }),
)

// To order the entire collection regardless of category
const allProductsByPrice = products.pipe(
  map((product) => [null, product]), // Key all products by null
  orderBy((product) => product.price, {
    comparator: (a, b) => a - b, // Sort by price ascending
  }),
)
```

#### 命令与索引

`orderByWithIndex(valueExtractor: (value: V) => T, options?: { comparator?: (a: T, b: T) => number, limit?: number, offset?: number })`

类似 `orderBy`,但结果中包括每个元素的位置索引。 输出格式为 `[key, [value, index]]` 其中指数是位置(从冲抵开始).

```typescript
// Order all products by price and include their overall rank
const rankedProducts = products.pipe(
  map((product) => [product.category, product]),
  orderByWithIndex((product) => product.price, {
    comparator: (a, b) => b - a, // Sort by price descending
    limit: 5,
  }),
  map(([category, [product, position]]) => ({
    category,
    product,
    rank: position + 1, // Convert to 1-based ranking
  })),
)
```

#### 命令By With Fractional Index

`orderByWithFractionalIndex(valueExtractor: (value: V) => T, options?: { comparator?: (a: T, b: T) => number, limit?: number, offset?: number })`

高级版本 `orderByWithIndex` 当元素移动位置时,使用分数索引以最小化变化。 这对UI应用程序特别有用,当命令改变时,您想要将 DOM 更新最小化 .

```typescript
// Order all products with stable fractional indices
const stableOrderedProducts = products.pipe(
  map((product) => [product.category, product]),
  orderByWithFractionalIndex((product) => product.rating, {
    comparator: (a, b) => b - a, // Sort by rating descending
  }),
  map(([category, [product, fractionalIndex]]) => ({
    category,
    product,
    position: fractionalIndex, // Lexicographically sortable string index
  })),
)
```

关于分数索引如何工作的更详细情况,参见: [执行分数索引](https://observablehq.com/@dgreensp/implementing-fractional-indexing) 大卫·格林斯潘著.

#### 顶端K

`topK(comparator: (a: T, b: T) => number, options?: { limit?: number, offset?: number })`

以比较国为基础限制结果的数目,可选限制和抵消。 这在按键流上起作用,其中元素在每一个按键群中进行排序.

```typescript
// Get the top 5 most expensive products in each category
const topProducts = products.pipe(
  map((product) => [product.category, product]), // Key by category
  topK((a, b) => b.price - a.price, { limit: 5 }), // Sort by price descending
)
```

要对整条流进行排序,所有元素按相同的值按键(例如 `null`:

```typescript
// Get the top 10 most expensive products overall
const topOverall = products.pipe(
  map((product) => [null, product]), // Key all products by null
  topK((a, b) => b.price - a.price, { limit: 10 }),
)
```

#### 顶端 KWith Index

`topKWithIndex(comparator: (a: T, b: T) => number, options?: { limit?: number, offset?: number })`

类似 `topK`,但结果中包括每个元素的位置索引。 输出格式为 `[key, [value, index]]` 其中指数是位置(从冲抵开始).

```typescript
// Get top 5 products with their position in each category
const rankedProducts = products.pipe(
  map((product) => [product.category, product]),
  topKWithIndex((a, b) => b.rating - a.rating, { limit: 5 }),
  map(([category, [product, position]]) => ({
    category,
    product,
    rank: position + 1, // Convert to 1-based ranking
  })),
)
```

#### 顶端 KWith Fractional Index 中

`topKWithFractionalIndex(comparator: (a: T, b: T) => number, options?: { limit?: number, offset?: number })`

高级版本 `topKWithIndex` 当元素移动位置时,使用分数索引以最小化变化。 它不使用整数指数,而是分配了以字符串为基础的分数指数,这些分数指数在词汇学上是可以分类的.

当元素改变位置时,只更新所移动元素的索引,而不是所有元素. 在您想要最小化 DOM 更新的 UI 应用程序中, 这特别有用 .

```typescript
// Get top 10 leaderboard entries with stable fractional indices
const leaderboard = scores.pipe(
  map((score) => ['leaderboard', score]),
  topKWithFractionalIndex((a, b) => b.points - a.points, { limit: 10 }),
  map(([_, [score, fractionalIndex]]) => ({
    ...score,
    position: fractionalIndex, // Lexicographically sortable string index
  })),
)
```

关于分数索引如何工作的更详细情况,参见: [执行分数索引](https://observablehq.com/@dgreensp/implementing-fractional-indexing) 大卫·格林斯潘著.

### 使用 SQLite 后端

对于持久性和较大的数据集,提供一些操作符,这些操作符持续使用SQLite:

-   `consolidate()`: 将数据合并为单一版本
-   `count()`: 计算收藏中的元素数
-   `distinct()`: 从收藏中删除重复
-   `join()`: 加入两个收藏
-   `map()`: 转换元素
-   `reduce()`: 按密钥汇总值
-   `groupBy()`: 按密钥分组数据并应用多个聚合函数
-   `topK()`: 根据比较国限制每组结果的数目及其变式 `topKWithIndex` 财务报告和审定财务报表 `topKWithFractionalIndex`
-   `orderBy()`:按数值排列的元素及其变体 `orderByWithIndex` 财务报告和审定财务报表 `orderByWithFractionalIndex`

每个都取 SQLite 数据库作为可选参数,例如:

```typescript
// Using better-sqlite3
const sqlite = new Database('./my_database.db')
const db = new BetterSQLite3Wrapper(sqlite)

const output = input.pipe(consolidate(db))
```

或者,您可以使用依赖性注射来提供一次SQLite数据库,用于操作器的管道:

```typescript
// Using better-sqlite3
const sqlite = new Database('./my_database.db')
const db = new BetterSQLite3Wrapper(sqlite)

// Using withSQLite for dependency injection
const output = input.pipe(
  withSQLite(db)(
    map((x) => x + 1),
    reduce((vals) => {
      let sum = 0
      for (const [val, diff] of vals) {
        sum += val * diff
      }
      return [[sum, 1]]
    }),
    distinct()
  )
)
```

这种方法使代码更清洁,无需将数据库单独传递给每个操作员。 这个 `withSQLite` 函数创建一个背景,使数据库在其范围内自动提供给所有SQLite操作员.

您也可以视需要混合两种方法:

```typescript
// Explicitly pass database to one operator
const firstPart = input.pipe(reduce(reducer, db))

// Use dependency injection for subsequent operators
firstPart.pipe(
  withSQLite(db)(
    distinct(),
    consolidate()
  )
)
```

运算符会自动创建必要的表格和索引来存储运算符的状态. 建议它对所有操作员使用相同的数据库,以确保状态存储在一个位置上.

这个 `BetterSQLite3Wrapper` 是一个包裹环绕 `better-sqlite3` 为操作员提供统一接口的库。 SQLite 其他数据库驱动程序可以通过执行 `SQLiteDb` 接口.

## 执行细节

此执行开始于 typeScript 移植 [将博客贴出内容化](https://materialize.com/blog/differential-from-scratch/),但有相当的分歧, 采用了管道的活字图案, 坚持到SQLite, 和一些其他的改变使 DX 更好.

1.  核心数据结构:
    
    -   `MultiSet`: 以倍数表示的收藏
    -   `Version`: 处理部分订购的版本
    -   `Antichain`: 管理边框
    -   `Index`: 存储版本操作员状态
2.  运算符 :
    
    -   基地操作员类别 `src/operators/`
    -   SQLite 变体 `src/sqlite/operators/`
3.  图执行 :
    
    -   数据流图管理 `src/graph.ts` 财务报告和审定财务报表 `src/D2.ts`
    -   在运算符之间传递信件
    -   边界跟踪和推进

## 参考资料

-   [差异数据流](https://github.com/MaterializeInc/differential)
-   [来自 Scratch 的不同数据流](https://materialize.com/blog/differential-from-scratch/)
-   [Python 执行](https://github.com/ruchirK/python-differential)
-   [DBSP 软件](https://arxiv.org/abs/2203.16684) (非常类似于差异数据流)

## 常见问题

### @electric-sql/d2ts 是什么？

D2TS是 Difficial Data flow 的一种类型脚本执行\.

### 如何安装 @electric-sql/d2ts？

在项目目录运行 npm install @electric-sql/d2ts，或使用 pnpm add / yarn add / bun add。

### @electric-sql/d2ts 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @electric-sql/d2ts 支持 ESM 还是 CommonJS？

以 ES Module 形式发布，推荐使用 import 引入。在 CommonJS 项目中，Node\.js 20\.19\+ / 22\.12\+ 可以直接 require\(\) 不含顶层 await 的 ES Module，更早的版本需要改用动态 import\(\)。

### @electric-sql/d2ts 可以商用吗？

@electric-sql/d2ts 使用 Apache-2\.0 许可。宽松许可：允许商业使用、修改和再分发，并附带明确的专利授权；分发时需保留许可与 NOTICE 文件，修改过的文件要注明改动。以上为概括，不构成法律意见，请以许可原文为准。

### @electric-sql/d2ts 的下载量有多少？

根据 npm 官方下载统计，@electric-sql/d2ts 在 2026/10/14 – 2026/11/12 这 30 天内被下载 11,487 次，日均约 383 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @electric-sql/d2ts 还在维护吗？

最新版本 0\.1\.8 发布于 2025/06/14（1\.4 年前）；至今共发布 9 个版本，其中过去 12 个月发布了 0 个；首次发布于 2025/02/27。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。