PHP 怎么用列存

wen PHP项目 2026-08-04 250

本文目录导读：

PHP 怎么用列存

使用列存数据库
使用 PHP 扩展
自己实现简单的列存
使用 row-oriented 转 column-oriented 的工具
现代数据库的列存支持
推荐方案

在 PHP 中使用列存（Columnar Storage）主要有以下几种方式，我会从简单到复杂介绍：

使用列存数据库

ClickHouse（推荐）

<?php
// 使用 ClickHouse 的 PHP 客户端
require_once 'vendor/autoload.php';
use ClickHouseDB\Client;
$config = [
    'host' => 'localhost',
    'port' => '8123',
    'username' => 'default',
    'password' => ''
];
$client = new Client($config);
// 创建列存表
$client->write('
    CREATE TABLE IF NOT EXISTS analytics.events (
        event_id UInt64,
        event_time DateTime,
        event_type String,
        user_id UInt64,
        data String
    ) ENGINE = MergeTree()
    ORDER BY (event_time)
');
// 批量插入数据
$client->insert(
    'events',
    [
        [1, '2024-01-01 10:00:00', 'click', 100, '{"page": "/home"}'],
        [2, '2024-01-01 10:01:00', 'view', 101, '{"page": "/about"}'],
        // ... 更多数据
    ],
    ['event_id', 'event_time', 'event_type', 'user_id', 'data']
);
// 列存高效的聚合查询
$result = $client->select('
    SELECT 
        event_type,
        COUNT(*) as cnt,
        AVG(user_id) as avg_user
    FROM events
    WHERE event_time >= today()
    GROUP BY event_type
');
print_r($result->rows());

Apache Cassandra

<?php
// 使用 DataStax PHP Driver
require_once 'vendor/autoload.php';
use Cassandra\Cluster;
use Cassandra\SimpleStatement;
$cluster = Cluster::builder()
    ->withContactPoints('127.0.0.1')
    ->build();
$session = $cluster->connect('analytics');
// 创建列族（表）
$session->execute(new SimpleStatement('
    CREATE TABLE IF NOT EXISTS user_events (
        user_id uuid,
        event_time timestamp,
        event_type text,
        PRIMARY KEY (user_id, event_time)
    ) WITH CLUSTERING ORDER BY (event_time DESC)
'));
// 插入数据
$prepared = $session->prepare(
    'INSERT INTO user_events (user_id, event_time, event_type) VALUES (?, ?, ?)'
);
$session->execute($prepared, [
    'arguments' => [
        new Cassandra\Uuid('12345678-1234-1234-1234-123456789012'),
        new Cassandra\Timestamp(time()),
        'click'
    ]
]);
// 查询
$result = $session->execute(new SimpleStatement('
    SELECT event_type, COUNT(*) as cnt
    FROM user_events
    WHERE user_id = ?
    GROUP BY event_type
'), ['arguments' => [new Cassandra\Uuid('12345678-1234-1234-1234-123456789012')]]);
foreach ($result as $row) {
    echo "Event: {$row['event_type']}, Count: {$row['cnt']}\n";
}

使用 PHP 扩展

Apache Arrow（内存列存格式）

<?php
// 需要安装 ext-arrow
$schema = new \Arrow\Schema([
    'id' => \Arrow\Type::INT64,
    'name' => \Arrow\Type::STRING,
    'score' => \Arrow\Type::FLOAT64
]);
$array = new \Arrow\RecordBatch($schema);
// 按列添加数据（列优先存储）
$array->addColumn('id', [1, 2, 3, 4, 5]);
$array->addColumn('name', ['Alice', 'Bob', 'Charlie', 'David', 'Eve']);
$array->addColumn('score', [85.5, 92.3, 78.9, 88.6, 95.1]);
// 列存的高效列操作
$scores = $array->column('score');
$average = array_sum($scores) / count($scores);
echo "Average score: $average\n";

自己实现简单的列存

<?php
class SimpleColumnStore {
    private $columns = [];
    private $rowCount = 0;
    // 添加列数据
    public function addColumn($name, array $data) {
        if ($this->rowCount === 0) {
            $this->rowCount = count($data);
        }
        if (count($data) !== $this->rowCount) {
            throw new Exception("Column length mismatch");
        }
        $this->columns[$name] = $data;
    }
    // 获取指定列（列存最优操作）
    public function getColumn($name) {
        return $this->columns[$name] ?? null;
    }
    // 聚合操作（列存高效场景）
    public function aggregate($column, $operation) {
        $data = $this->getColumn($column);
        if (!$data) return null;
        switch ($operation) {
            case 'sum':
                return array_sum($data);
            case 'avg':
                return array_sum($data) / count($data);
            case 'min':
                return min($data);
            case 'max':
                return max($data);
            default:
                return null;
        }
    }
    // 条件筛选（列存劣势场景）
    public function filter($column, $callback) {
        $indices = array_keys(array_filter($this->columns[$column], $callback));
        $result = new self();
        foreach ($this->columns as $name => $data) {
            $result->addColumn($name, array_intersect_key($data, array_flip($indices)));
        }
        return $result;
    }
}
// 使用示例
$store = new SimpleColumnStore();
$store->addColumn('id', range(1, 1000000));
$store->addColumn('value', range(100, 1000000 + 99));
// 高效的聚合操作
echo $store->aggregate('value', 'avg'); // 500000.5

使用 row-oriented 转 column-oriented 的工具

<?php
// 将行式数据转换为列式存储
class RowToColumnTransformer {
    public static function transform(array $rows) {
        $columns = [];
        foreach ($rows as $row) {
            foreach ($row as $key => $value) {
                $columns[$key][] = $value;
            }
        }
        return $columns;
    }
}
// 使用
$data = [
    ['id' => 1, 'name' => 'Alice', 'age' => 25],
    ['id' => 2, 'name' => 'Bob', 'age' => 30],
    ['id' => 3, 'name' => 'Charlie', 'age' => 35],
];
$columnar = RowToColumnTransformer::transform($data);
print_r($columnar);
// 输出：['id' => [1,2,3], 'name' => ['Alice','Bob','Charlie'], 'age' => [25,30,35]]

现代数据库的列存支持

<?php
// MySQL 8.0+ 不原生支持，但可以通过临时表模拟
// PostgreSQL 12+ 的列存扩展
$pdo = new PDO('pgsql:host=localhost;dbname=test', 'user', 'pass');
// 创建列存表（使用 cstore_fdw 扩展）
$pdo->exec("CREATE EXTENSION IF NOT EXISTS cstore_fdw");
$pdo->exec("CREATE SERVER cstore_server FOREIGN DATA WRAPPER cstore_fdw");
$pdo->exec("CREATE FOREIGN TABLE events (
    id integer,
    timestamp timestamp,
    event_type text
) SERVER cstore_server OPTIONS(compression 'pglz')");
// 插入数据（会自动转换为列存）
$stmt = $pdo->prepare("INSERT INTO events VALUES (?, ?, ?)");
$stmt->execute([1, time(), 'click']);
// 列存的聚合查询
$result = $pdo->query("
    SELECT event_type, COUNT(*) 
    FROM events 
    GROUP BY event_type
");