
![截屏2025-11-11 15.53.23.png](/static/img/54771b327d9630c3a55ad75596f9ea88.%C3%A6%C2%88%C2%AA%C3%A5%C2%B1%C2%8F2025-11-11%2015.53.23.webp)

2025/11/11

# Substreams 流式框架入门：Solana 链上数据实时处理

> 本文介绍 Substreams 流式框架的核心概念，以 Solana 生态为例，讲解 WASM 运行时、Protobuf 序列化、模块化架构等关键技术。

## 背景

在区块链数据处理领域，传统的索引方案往往面临以下挑战：

- **延迟高**：轮询式抓取无法做到实时响应
- - **扩展难**：新增协议需要大量重构
- - **性能瓶颈**：单线程处理难以应对高吞吐

[Substreams](https://substreams.streamingfast.io/) 是 StreamingFast 推出的流式数据处理框架，专为区块链数据索引而设计。本文将以 Solana 生态为例，介绍其核心原理与实践。

---

## 核心概念

### 1. 流式处理模型

Substreams 采用**推送模型**，由节点主动推送区块数据，而非客户端轮询：

```
传统模式: Client -> Request -> Node -> Response (轮询)
Substreams: Node -> Block Stream -> Client (推送)
```

这种模式的优势：
- **低延迟**：新区块产生即刻推送
- - **高效利用**：避免无效轮询请求
- - **断点续传**：通过游标(cursor)实现精确恢复

### 2. WebAssembly 运行时

Substreams 的数据处理逻辑编译为 **WebAssembly (WASM)**，在沙箱环境中执行：

\`\`\`bash
# 编译为 WASM
cargo build --target wasm32-unknown-unknown --release
\`\`\`

WASM 的优势：
- **安全性**：沙箱隔离，防止恶意代码
- - **跨平台**：一次编译，多处运行
- - **高性能**：接近原生的执行速度
- - **可移植**：模块可分发复用

### 3. Protocol Buffers 序列化

数据格式采用 **Protocol Buffers**，相比 JSON 有显著优势：

| 特性 | Protobuf | JSON |
|------|----------|------|
| 序列化大小 | 小（二进制） | 大（文本） |
| 解析速度 | 快 | 慢 |
| Schema 强类型 | 是 | 否 |
| 向后兼容 | 支持 | 需手动处理 |

---

## 模块化架构设计

以 [solana-substreams](https://github.com/0xpapercut/solana-substreams) 项目为例，展示模块化架构：

\`\`\`
.
├── solana_hub/          # 核心汇聚模块
├── spl_token/           # SPL Token 事件处理
├── system_program/      # 系统程序事件
├── raydium_amm/         # Raydium AMM 协议
├── pumpfun/             # Pumpfun Meme 币平台
├── meteora_dlmm/        # Meteora DLMM 协议
└── orca_whirlpools/     # Orca 集中流动性协议
\`\`\`

### 模块独立性

每个模块独立封装，可单独运行：

\`\`\`bash
# 单独运行 Raydium 模块
cd raydium_amm && substreams gui

# 单独运行 Pumpfun 模块
cd pumpfun && substreams gui
\`\`\`

### 统一汇聚层

\`solana_hub\` 作为汇聚层，整合所有子模块数据流。

---

## 实战：游标持久化

在流式处理中，**游标(cursor)** 标记当前处理位置，用于断点续传。

### 批量缓存策略

达到 10000 个游标时批量提交，每小时强制刷新。

### Redis Pipeline 持久化

使用 Redis Pipeline 一次网络往返完成所有写入。

---

## 构建与部署

### 环境准备

\`\`\`bash
# 安装 Substreams CLI
curl https://substreams.streamingfast.io/install | bash

# 设置 API 密钥
export STREAMINGFAST_KEY=your_api_key_here

# 安装 Rust target
rustup target add wasm32-unknown-unknown
\`\`\`

### 构建流程

\`\`\`bash
# 生成 Protobuf 代码
buf generate

# 编译为 WASM
make build

# 打包为 Substreams 包
make package

# 启动数据流
make stream START=344531182
\`\`\`

---

## 性能优化要点

### 1. 批量处理

- 游标批量持久化（减少 Redis 网络往返）
- - Kafka 批量投递（提高吞吐量）

### 2. 并发模型

- 使用 Tokio 异步运行时
- - 多模块并行处理

### 3. 内存管理

- 避免不必要的克隆
- - 使用引用计数智能指针

---

## 应用场景

1. **DeFi 数据索引**：实时捕获交易、流动性变化
2. 2. **NFT 追踪**：监控铸造、转移事件
3. 3. **链上监控**：特定地址或合约活动追踪
4. 4. **数据分析**：构建实时数据仓库

---

## 总结

Substreams 为区块链数据处理提供了现代化的解决方案：

- **流式推送**：低延迟实时处理
- - **WASM 运行时**：安全、高效、可移植
- - **Protobuf 序列化**：紧凑、快速、强类型
- - **模块化架构**：易扩展、可复用

对于 Solana 生态的开发者，Substreams 是构建高性能数据基础设施的理想选择。

---

## 参考资料

- [Substreams 官方文档](https://substreams.streamingfast.io/)
- - [Solana 程序开发文档](https://docs.solana.com/)
- - [Protocol Buffers 文档](https://developers.google.com/protocol-buffers)
- - [solana-substreams 项目](https://github.com/0xpapercut/solana-substreams)

Substreams 流式框架入门：Solana 链上数据实时处理

# ClickHouse 列式存储：大数据分析的性能优化实践

## 背景

在大数据分析领域，传统的行式数据库在面对海量数据查询时往往力不从心。ClickHouse 作为一款高性能的列式数据库管理系统（DBMS），专为在线分析处理（OLAP）场景设计，以其惊人的查询性能和高效的数据压缩能力，成为大数据分析领域的热门选择。

## 列式存储 vs 行式存储

### 行式存储

传统数据库（如 MySQL、PostgreSQL）采用行式存储，数据按行组织：

```
| ID | Name  | Age | City    |
|----|-------|-----|---------|
| 1  | Alice | 25  | Beijing |
| 2  | Bob   | 30  | Shanghai|
| 3  | Carol | 28  | Guangzhou|
```

存储方式：`1|Alice|25|Beijing|2|Bob|30|Shanghai|3|Carol|28|Guangzhou`

### 列式存储

ClickHouse 采用列式存储，数据按列组织：

```
ID列:   1|2|3
Name列: Alice|Bob|Carol
Age列:  25|30|28
City列: Beijing|Shanghai|Guangzhou
```

### 列式存储的优势

1. **查询性能提升**
   - 只读取需要的列，减少 I/O
   - 聚合查询（SUM、COUNT、AVG）性能显著提升

2. **高压缩比**
   - 同列数据类型相同，压缩效率高
   - ClickHouse 支持多种压缩算法（LZ4、ZSTD、Delta等）

3. **向量化执行**
   - 利用 SIMD 指令集并行处理数据
   - 批量处理提升 CPU 利用率

## ClickHouse 核心特性

### 1. 表引擎

ClickHouse 提供多种表引擎，适应不同场景：

```sql
-- MergeTree 系列：核心引擎，支持主键、分区、索引
CREATE TABLE events (
    event_date Date,
    user_id UInt64,
    event_type String,
    event_data String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, user_id);

-- ReplacingMergeTree：自动去重
CREATE TABLE users (
    user_id UInt64,
    name String,
    updated_at DateTime
) ENGINE = ReplacingMergeTree(updated_at)
ORDER BY user_id;

-- SummingMergeTree：预聚合
CREATE TABLE stats (
    date Date,
    metric String,
    value UInt64
) ENGINE = SummingMergeTree()
ORDER BY (date, metric);

-- Distributed：分布式表
CREATE TABLE events_all AS events
ENGINE = Distributed(cluster_name, database, events, rand());
```

### 2. 分区与索引

```sql
-- 分区策略：按月分区
PARTITION BY toYYYYMM(event_date)

-- 主键排序
ORDER BY (event_date, user_id)

-- 二级索引
CREATE TABLE events (
    event_date Date,
    user_id UInt64,
    event_type String,
    INDEX idx_type event_type TYPE bloom_filter GRANULARITY 4
) ENGINE = MergeTree()
ORDER BY (event_date, user_id);
```

### 3. 数据跳过索引

```sql
-- Bloom Filter 索引
INDEX idx_bloom column_name TYPE bloom_filter GRANULARITY 1

-- MinMax 索引
INDEX idx_minmax column_name TYPE minmax GRANULARITY 3

-- Set 索引
INDEX idx_set column_name TYPE set(0) GRANULARITY 1
```

## 性能优化实践

### 1. 写入优化

```sql
-- 批量写入（推荐每次 10000-100000 行）
INSERT INTO events VALUES
    ('2025-11-01', 1, 'click', 'data1'),
    ('2025-11-01', 2, 'view', 'data2'),
    ...;

-- 避免小批量频繁写入
-- 不推荐：每次插入几行
-- 推荐：累积后批量插入
```

### 2. 查询优化

```sql
-- 使用 PREWHERE 替代 WHERE（只读取需要的列）
SELECT user_id, count()
FROM events
PREWHERE event_date = '2025-11-01'
WHERE event_type = 'click'
GROUP BY user_id;

-- 分区裁剪
SELECT count()
FROM events
WHERE event_date BETWEEN '2025-11-01' AND '2025-11-30';  -- 只扫描11月分区

-- 避免使用 SELECT *
SELECT user_id, event_type FROM events;  -- 好
SELECT * FROM events;  -- 避免
```

### 3. 内存优化

```sql
-- 限制内存使用
SET max_memory_usage = 10000000000;  -- 10GB

-- 使用外部排序
SET max_bytes_before_external_group_by = 20000000000;
```

## 集群部署

### 配置示例

```xml

<remote_servers>
    <cluster_name>
        <shard>
            <replica>
                <host>node1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>node2</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <replica>
                <host>node3</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>node4</host>
                <port>9000</port>
            </replica>
        </shard>
    </cluster_name>
</remote_servers>
```

### 分布式查询

```sql
-- 创建本地表
CREATE TABLE events_local ON CLUSTER cluster_name (
    event_date Date,
    user_id UInt64,
    event_type String
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, user_id);

-- 创建分布式表
CREATE TABLE events_all ON CLUSTER cluster_name AS events_local
ENGINE = Distributed(cluster_name, default, events_local, rand());

-- 查询分布式表
SELECT count() FROM events_all WHERE event_date = today();
```

## 实际应用场景

### 1. 日志分析

```sql
-- 创建日志表
CREATE TABLE logs (
    timestamp DateTime,
    level String,
    message String,
    source String,
    INDEX idx_level level TYPE bloom_filter GRANULARITY 2
) ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (timestamp, level);

-- 查询错误日志统计
SELECT 
    toStartOfHour(timestamp) as hour,
    level,
    count() as count
FROM logs
WHERE timestamp >= now() - INTERVAL 24 HOUR
GROUP BY hour, level
ORDER BY hour, level;
```

### 2. 用户行为分析

```sql
-- 用户行为表
CREATE TABLE user_events (
    event_date Date,
    user_id UInt64,
    event_type String,
    page_id UInt64,
    duration UInt32
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, user_id, event_type);

-- 漏斗分析
SELECT 
    user_id,
    countIf(event_type = 'view') as views,
    countIf(event_type = 'click') as clicks,
    countIf(event_type = 'purchase') as purchases
FROM user_events
WHERE event_date = today()
GROUP BY user_id
HAVING views > 0;
```

### 3. 实时监控

```sql
-- 物化视图实现实时聚合
CREATE MATERIALIZED VIEW stats_mv
ENGINE = SummingMergeTree()
ORDER BY (date, metric)
AS SELECT
    toDate(timestamp) as date,
    metric,
    count() as value
FROM metrics
GROUP BY date, metric;

-- 查询实时统计
SELECT * FROM stats_mv WHERE date = today();
```

## 与其他系统集成

### Kafka 集成

```sql
-- 创建 Kafka 引擎表
CREATE TABLE kafka_queue (
    timestamp DateTime,
    level String,
    message String
) ENGINE = Kafka()
SETTINGS 
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'logs',
    kafka_group_name = 'clickhouse',
    kafka_format = 'JSONEachRow';

-- 创建物化视图消费数据
CREATE MATERIALIZED VIEW kafka_consumer TO logs AS
SELECT * FROM kafka_queue;
```

### Go 客户端示例

```go
package main

import (
    "context"
    "fmt"
    "time"

    "github.com/ClickHouse/clickhouse-go/v2"
)

func main() {
    conn, err := clickhouse.Open(&clickhouse.Options{
        Addr: []string{"127.0.0.1:9000"},
        Auth: clickhouse.Auth{
            Database: "default",
            Username: "default",
            Password: "",
        },
        Settings: clickhouse.Settings{
            "max_execution_time": 60,
        },
    })
    if err != nil {
        panic(err)
    }
    defer conn.Close()

    // 批量插入
    batch, err := conn.PrepareBatch(context.Background(), 
        "INSERT INTO events (event_date, user_id, event_type)")
    if err != nil {
        panic(err)
    }

    for i := 0; i < 10000; i++ {
        batch.Append(
            time.Now(),
            uint64(i),
            "click",
        )
    }
    batch.Send()

    // 查询
    rows, err := conn.Query(context.Background(),
        "SELECT user_id, count() FROM events GROUP BY user_id")
    if err != nil {
        panic(err)
    }
    defer rows.Close()

    for rows.Next() {
        var userID uint64
        var count uint64
        rows.Scan(&userID, &count)
        fmt.Printf("User: %d, Count: %d\n", userID, count)
    }
}
```

## 性能对比

| 场景 | MySQL | ClickHouse | 提升 |
|------|-------|------------|------|
| 亿级数据聚合查询 | 30s | 0.5s | 60x |
| 日志分析（10亿行） | 5min | 3s | 100x |
| 用户行为分析 | 10s | 0.2s | 50x |
| 数据压缩比 | 1:1 | 1:10 | 10x |

## 最佳实践总结

1. **数据建模**
   - 合理选择分区键（通常为日期）
   - 优化排序键，提高查询性能
   - 使用合适的表引擎

2. **写入策略**
   - 批量写入，避免小批量频繁写入
   - 控制写入并发度
   - 使用 Buffer 表缓冲写入

3. **查询优化**
   - 使用 PREWHERE 过滤
   - 避免分布式 JOIN
   - 合理使用物化视图

4. **集群运维**
   - 合理规划分片和副本
   - 监控 ZooKeeper 状态
   - 定期清理过期数据

## 总结

ClickHouse 的列式存储架构使其在大数据分析场景中展现出卓越的性能。通过合理的表设计、查询优化和集群部署，可以构建高效的数据分析平台。在海量数据处理、实时分析、日志监控等场景中，ClickHouse 是一个值得考虑的优秀选择。

---

> 本文介绍了 ClickHouse 列式存储的核心原理和性能优化实践，包括表引擎选择、分区索引策略、集群部署方案以及实际应用案例。

ClickHouse 列式存储：大数据分析的性能优化实践

前几天接手研究生师兄的flutter项目，需要切换低版本的flutter,于是下载了fvm，下载版本后切换一直卡在

```
PS C:\Users\zyh16> fvm install 2.10.5
✓ Flutter SDK: SDK Version : 2.10.5 installed! (17.2s)
PS C:\Users\zyh16> fvm use 2.10.5

No pubspec.yaml detected in this directory
✔ Would you like to continue? · yes                                                                                    
Setting up Flutter SDK: 2.10.5

Checking Dart SDK version...
Downloading Dart SDK from Flutter engine ...
Expanding downloaded archive...
Building flutter tool...
Running pub upgrade...
```
更换代理也没用，临时环境变量也不生效

然后添加系统环境变量后解决
```
PUB_HOSTED_URL https://pub.flutter-io.cn

FLUTTER_STORAGE_BASE_URL https://storage.flutter-io.cn
```

fvm切换版本一直卡在Running pub upgrade...

之前有两个项目都有巡查功能，基于电子围栏，获取两个点的距离是很常见的功能，用来统计当前用户到目标范围中心点大致距离，接受一定的误差,使用Haversine 公式
就能实现

```java


import cn.hutool.core.util.NumberUtil;



public class LocationUtils {

    private static double EARTH_RADIUS = 6378.137;

    private static double rad(double d) {
        return d * Math.PI / 180.0;
    }



    /**
     * 通过经纬度获取距离(单位：千米)
     *
     * @param lat1
     * @param lng1
     * @param lat2
     * @param lng2
     * @return
     */
    public static double getDistance(double lat1, double lng1, double lat2,
                                     double lng2) {
        double radLat1 = rad(lat1);
        double radLat2 = rad(lat2);
        double a = radLat1 - radLat2;
        double b = rad(lng1) - rad(lng2);
        double s = 2 * Math.asin(Math.sqrt(Math.pow(Math.sin(a / 2), 2)
                + Math.cos(radLat1) * Math.cos(radLat2)
                * Math.pow(Math.sin(b / 2), 2)));
        s = s * EARTH_RADIUS;
        s = Math.round(s * 10000d) / 10000d;
        return NumberUtil.round(s, 2).doubleValue();
    }

}
```
这是一个很常见的实现

下面给出其他语言版本
```golang
const earthRadius = 6371 // 地球平均半径，单位：千米

// degToRad 将角度转换为弧度
func degToRad(deg float64) float64 {
	return deg * math.Pi / 180
}

// GetDistance 计算两个经纬度之间的距离（单位：千米）
func GetDistance(lat1, lng1, lat2, lng2 float64) float64 {
	radLat1 := degToRad(lat1)
	radLat2 := degToRad(lat2)
	a := radLat1 - radLat2
	b := degToRad(lng1) - degToRad(lng2)
	s := 2 * math.Asin(math.Sqrt(math.Pow(math.Sin(a/2), 2) +
		math.Cos(radLat1)*math.Cos(radLat2)*math.Pow(math.Sin(b/2), 2)))
	s *= earthRadius
	s = math.Round(s*100) / 100 // 保留两位小数
	return s
}

```

Js版本
```js
const earthRadius = 6371; // 地球平均半径，单位：千米

function degToRad(deg) {
  return deg * Math.PI / 180;
}

function getDistance(lat1, lng1, lat2, lng2) {
  const radLat1 = degToRad(lat1);
  const radLat2 = degToRad(lat2);
  const a = radLat1 - radLat2;
  const b = degToRad(lng1) - degToRad(lng2);
  let s = 2 * Math.asin(Math.sqrt(Math.pow(Math.sin(a / 2), 2) +
    Math.cos(radLat1) * Math.cos(radLat2) * Math.pow(Math.sin(b / 2), 2)));
  s *= earthRadius;
  s = Math.round(s * 100) / 100; // 保留两位小数
  return s;
}
```