3、数据导入与导出：从CSV/JSON导入数据、从MySQL/PostgreSQL同步、导出到外部系统

数据导入导出，说白了就是让 ClickHouse 跟外界「打交道」。我刚开始用 ClickHouse 时，觉得它就是个查询引擎，后来才发现，数据怎么进去、怎么出来，才是真正考验功力的地方。你想想看，一个系统再快，数据进不来或者出不去，那都是白搭。

这一章，咱们就聊聊数据怎么从 CSV、JSON 这些常见格式进来，怎么从 MySQL、PostgreSQL 实时同步，以及怎么把处理好的数据导出到外部系统。嗯，这里要注意，每种方式都有它的适用场景，选错了可是要吃苦头的。

3.1 从 CSV/JSON 导入数据

CSV 和 JSON 是最常见的两种数据交换格式。我个人习惯用 CSV 做批量导入，用 JSON 做实时流式接入。为什么？因为 CSV 结构简单，解析快；JSON 虽然灵活，但解析开销大，适合小批量。

3.1.1 使用 INSERT 语句导入 CSV

最直接的方式就是用 INSERT INTO ... FORMAT CSV。我在项目中遇到过，很多新手会用 INSERT INTO table VALUES (...) 一条一条插，那效率简直惨不忍睹。正确的做法是：

-- 从本地文件导入
cat data.csv | clickhouse-client --query="INSERT INTO my_table FORMAT CSV"

-- 或者直接在客户端里
clickhouse-client --query="INSERT INTO my_table FORMAT CSV" < data.csv

这里有个坑：CSV 文件里的引号、转义符处理。我曾经因为 CSV 里某个字段包含逗号，结果数据全乱了。解决办法是：

• 字段用双引号包裹
• 内部双引号用两个双引号转义
• 或者用 FORMAT CSVWithNames 带上表头

3.1.2 使用 INSERT 语句导入 JSON

JSON 导入稍微复杂点。ClickHouse 支持 FORMAT JSONEachRow，也就是每行一个 JSON 对象。这种格式我最喜欢，因为它天然支持嵌套结构。

-- 示例 JSON 数据
{"id": 1, "name": "Alice", "score": 95.5}
{"id": 2, "name": "Bob", "score": 87.3}

-- 导入命令
cat data.json | clickhouse-client --query="INSERT INTO my_table FORMAT JSONEachRow"

但要注意，JSON 里的字段名必须和表结构完全一致，大小写敏感。我踩过这个坑：表里字段是 user_name，JSON 里写的是 UserName，结果导入后全是 NULL。

3.1.3 使用表函数导入

ClickHouse 提供了 file() 和 url() 表函数，可以直接读取外部文件或 HTTP 接口的数据。我个人觉得这个功能特别适合做临时数据探索。

-- 从本地文件读取 CSV
SELECT * FROM file('data.csv', 'CSV', 'id UInt32, name String, score Float32')

-- 从 URL 读取 JSON
SELECT * FROM url('http://example.com/data.json', 'JSONEachRow', 'id UInt32, name String')

3.2 从 MySQL/PostgreSQL 同步

实际业务中，数据往往存在关系型数据库里。怎么把这些数据实时同步到 ClickHouse？我试过好几种方案，这里说说最靠谱的两种。

3.2.1 使用 MaterializedMySQL 引擎

ClickHouse 21.3 之后提供了 MaterializedMySQL 数据库引擎，可以实时同步 MySQL 数据。说白了，就是 ClickHouse 里建一个「影子库」，MySQL 那边一有变化，这边自动更新。

-- 创建 MaterializedMySQL 数据库
CREATE DATABASE mysql_db ENGINE = MaterializedMySQL(
    'localhost:3306',
    'source_db',
    'user',
    'password'
)

-- 然后就可以直接查询了
SELECT * FROM mysql_db.orders WHERE order_date > '2024-01-01'

嗯，这里要注意几个坑：

• MySQL 表必须有主键，否则同步会失败
• 不支持 DDL 变更（比如加字段），需要手动重建
• 大表首次同步会很慢，我试过 1 亿行的表，同步了将近 2 小时

3.2.2 使用 PostgreSQL 的 Logical Replication

PostgreSQL 的同步方式不太一样。ClickHouse 提供了 PostgreSQL 表引擎，可以直接查询 PG 表，但这不是实时同步。真正实时同步需要借助 PG 的逻辑复制功能。

-- 创建 PostgreSQL 表引擎
CREATE TABLE pg_orders (
    id UInt32,
    customer_name String,
    amount Float32
) ENGINE = PostgreSQL('localhost:5432', 'db', 'orders', 'user', 'password')

但说实话，这种方式的性能一般。我在项目中更推荐用 Debezium + Kafka 的方案：PG 的 WAL 日志通过 Debezium 捕获，发到 Kafka，然后 ClickHouse 从 Kafka 消费。虽然架构复杂点，但稳定性和性能都好很多。

3.3 导出到外部系统

数据进来处理完了，总得出去吧？ClickHouse 的导出能力其实很强，只是很多人不知道。

3.3.1 导出为 CSV/JSON

最常用的就是导出为 CSV 或 JSON 文件。我一般用 INTO OUTFILE 语句：

-- 导出为 CSV
SELECT * FROM my_table
INTO OUTFILE '/tmp/export.csv'
FORMAT CSV

-- 导出为 JSON
SELECT * FROM my_table
INTO OUTFILE '/tmp/export.json'
FORMAT JSONEachRow

这里有个细节：导出路径必须是 ClickHouse 服务器能访问的路径，不是客户端路径。我曾经在本地跑这个命令，结果文件写到了服务器上，找了半天没找到。

3.3.2 导出到 MySQL/PostgreSQL

ClickHouse 没有直接导出到 MySQL 或 PG 的内置功能。我的做法是：

1. 先用 INTO OUTFILE 导出为 CSV
2. 再用 MySQL 的 LOAD DATA INFILE 或 PG 的 COPY 命令导入

或者，如果你用 Python，可以写个脚本：

import clickhouse_driver
import pymysql

# 从 ClickHouse 查询
client = clickhouse_driver.Client(host='localhost')
rows = client.execute('SELECT * FROM my_table')

# 写入 MySQL
conn = pymysql.connect(host='localhost', user='root', password='123456', database='target_db')
cursor = conn.cursor()
cursor.executemany('INSERT INTO target_table VALUES (%s, %s, %s)', rows)
conn.commit()

3.3.3 使用表函数导出

ClickHouse 的 url() 表函数也可以用来导出数据。比如把数据 POST 到一个 HTTP 接口：

INSERT INTO FUNCTION
url('http://api.example.com/ingest', 'JSONEachRow')
SELECT * FROM my_table WHERE date = today()

这个功能我用的不多，因为网络延迟和重试机制不太好控制。但如果你有现成的 HTTP 接收端，倒是挺方便的。

3.4 本章知识体系

下面这张图总结了数据导入导出的核心逻辑，我画的时候特意把「实时同步」和「批量导入」分开，因为它们的架构思路完全不同。

好了，这一章就聊到这儿。数据怎么进来、怎么出去，心里有数了吧？

---