ByteHouse 云数仓版支持通过 ClickHouse 原生 NodeJS 客户端连接。本文介绍了如何使用 ClickHouse 连接 ByteHouse 云数仓版并执行建表、查询等操作。

请确保您使用的 ClickHouse NodeJS 客户端版本为 v2.4.2 及以上版本。

• 暂时不支持 ByteHouse 的 JSONB、BitMap64 类型。
• 如果您在使用过程中遇到其他未知限制，请联系 ByteHouse 团队处理。

使用 npm 安装 ClickHouse NodeJS 客户端。

npm install clickhouse@2.4.2

ByteHouse 支持通过 IAM 用户或数据库用户连接 ClickHouse NodeJS。IAM 用户与数据库用户二者差异说明如下，您可按需选择。

• IAM 用户为火山引擎访问控制（IAM）中创建的用户，其权限由 IAM 权限策略及您授予的 ByteHouse 资源和数据权限决定。IAM 用户可访问 ByteHouse 控制台，也支持通过 CLI、连接驱动、生态工具、API 等方式访问 ByteHouse。
• 数据库用户为 ByteHouse 中创建的数据库级别用户，可为其授予环境、资源和数据权限。数据库用户不可访问 ByteHouse 控制台，但支持通过 CLI、连接驱动、生态工具、API 等方式访问 ByteHouse。

更多 IAM 用户和数据库用户的介绍请参见以下文档：

• ByteHouse权限管控概述
• 管理 IAM 用户
• 管理数据库用户

使用 IAM 用户连接

请参考步骤三：获取 ByteHouse 连接串信息，了解如何通过 IAM 用户方式连接到 ByteHouse。
通用参数说明如下：

使用数据库用户连接

连接 ByteHouse

可参考下面代码连接至 ByteHouse，使用时注意替换连接语句中的 {host} 、{password}、{user}、{Database}、{VIRTUAL_WAREHOUSE_ID} 等连接信息字段，获取方式请参见获取 ByteHouse 连接信息。

const host = '{Host}';
const port = 8123;
const password = '{Password}';
const user = '{User}';
const database = '{Database}';
const virtual_warehouse = '{VIRTUAL_WAREHOUSE_ID}';

const clickhouse = new ClickHouse({
    url: util.format('https://%s', host),
    port: port,
    debug: false,

    basicAuth: {
        username: user,
        password: password,
    },

    isUseGzip: false,
    usePost: false,
    format: 'json',

    config: {
        session_timeout: 60,
        output_format_json_quote_64bit_integers: 0,
        enable_http_compression: 0,
        database: database,
        virtual_warehouse: virtual_warehouse,
    },

});

参数说明如下：

数据库基础操作

连接成功后，您可使用以下代码创建数据库、表、插入数据并查询数据。

import { ClickHouse } from 'clickhouse';
import * as util from "node:util";

const host = '{Host}';
const port = 8123;
const password = '{Password}';
const user = '{User}';
const database = '{Database}';
const virtual_warehouse = '{VIRTUAL_WAREHOUSE_ID}';

const clickhouse = new ClickHouse({
    url: util.format('https://%s', host),
    port: port,
    debug: false,

    basicAuth: {
        username: user,
        password: password,
    },

    isUseGzip: false,
    usePost: false,
    format: 'json',

    config: {
        session_timeout: 60,
        output_format_json_quote_64bit_integers: 0,
        enable_http_compression: 0,
        database: database,
        virtual_warehouse: virtual_warehouse,
    },

});

// @ts-ignore
// 完整案例
async function run() {
    try {
        await clickhouse.query('DROP DATABASE IF EXISTS bhnodetest').toPromise();
        await clickhouse.query('CREATE DATABASE IF NOT EXISTS bhnodetest').toPromise();
        await clickhouse.query(`
      CREATE TABLE bhnodetest.sample_table (
        id Int32,
        name String
      ) ENGINE = CnchMergeTree()
      ORDER BY id
    `).toPromise();

        await clickhouse.insert(
            'INSERT INTO bhnodetest.sample_table (id, name)',
            [
                [1, 'Alice'],
                [2, 'Bob'],
            ]
        ).toPromise();

        const rows = await clickhouse.query('SELECT * FROM bhnodetest.sample_table FORMAT JSON').toPromise();

        console.log('Query result:', rows);
    } catch (err) {
        console.error('Error running ClickHouse operations:', err);
    }
}

// @ts-ignore
await run();

使用回调函数查询

该示例演示了通过回调函数处理查询结果的异步执行方式。

clickhouse.query('SELECT * FROM bhnodetest.sample_table FORMAT JSON').exec((err, rows) => {
    if (err) {
        console.error('Callback query error:', err);
    } else {
        console.log('Query result (Callback):', rows);
    }
});

流式查询

该示例演示了通过流（Stream）方式处理查询结果，适用于返回数据量较大的场景，如批量读取数据，可逐行处理数据而非等待全部加载完成。

clickhouse.query('SELECT number FROM system.numbers LIMIT 5 FORMAT JSON').stream()
  .on('data', function (chunk) {
    const stream = this;
    process.stdout.write('Stream chunk: ' + chunk.toString());
    stream.pause();
    setTimeout(() => stream.resume(), 1000); // 模拟背压场景
  })
  .on('error', err => {
    console.error('Stream error:', err);
  })
  .on('end', () => {
    console.log('\nStream ended.');
  });

流式插入

该示例演示了通过流（Stream）方式处理查询结果，适用于返回数据量较大的场景，如批量读取数据，可逐行处理数据而非等待全部加载完成。

clickhouse.query('SELECT number FROM system.numbers LIMIT 5 FORMAT JSON').stream()
    .on('data', function (chunk) {
        const stream = this;
        process.stdout.write('Stream chunk: ' + chunk.toString());
        stream.pause();
        setTimeout(() => stream.resume(), 1000); // 模拟背压场景
    })
    .on('error', err => {
        console.error('Stream error:', err);
    })
    .on('end', () => {
        console.log('\nStream ended.');
    });