用 ClickHouse 将 Parquet 数据加载到 MySQL

几周前，我正在为ClickPipes 的 MySQL CDC 连接器制作一个视频，我需要将一些 Parquet 文件中的数据导入 MySQL。没有原生的方法可以做到这一点，所以我决定使用clickhouse-local，这是我最喜欢的用于这类临时数据工作的工具。

设置 MySQL

在视频中我使用了 AWS RDS for MySQL，但为了更容易复现，在这篇博客文章中我们将使用运行在我机器上的 MySQL。我们可以通过运行以下命令来启动 MySQL：

1docker run --name parquet-mysql \2    -p 127.0.0.1:3306:3306 \3    -e MYSQL_ROOT_PASSWORD=local-root-password \4    -e MYSQL_DATABASE=stackoverflow \5    -e MYSQL_USER=admin \6    -e MYSQL_PASSWORD=local-demo-password \7    -v parquet-mysql-data:/var/lib/mysql \8    mysql:8.4

设置 ClickHouse

我们将通过其二进制文件运行 ClickHouse，可以像这样下载：

1curl https://clickhouse.com | sh

然后我们将启动它：

1./clickhouse -mn \2--config-file ch-config.yaml \3--param_mysql_password "local-demo-password" \4--param_mysql_host "127.0.0.1:3306" \5--output-format Pretty

提示

我们以明文形式传入 MySQL 主机和密码，因为它运行在我们自己的机器上。对于生产系统，我们会将这些值存储在环境变量中。

其内容为 ch-config.yaml 如下所示：

1echo:02users_config:ch-users.yaml3named_collections:4mysql_demo:5host:'127.0.0.01'6port:33067user:'admin'8password:'local-demo-password'9database:'stackoverflow'

ch-users.yaml 包含默认用户的配置。命名集合 mysql_demo 包含我们稍后将在文章中使用的 MySQL 凭据。

探索 StackOverflow 数据集

现在我们已经运行了 MySQL 和 ClickHouse，让我们来探索我们的数据集。我们将使用 StackOverflow 投票数据，它位于以下 S3 存储桶中：

1SET url_base ='https://datasets-documentation.s3.eu-west-3.amazonaws.com/stackoverflow/parquet/votes/';

在使用 url 表函数时，url_base 参数会被添加到相对路径前面，这样我们就不必多次重复相同的基本 URL。我们可以像这样描述该存储桶中的 2024.parquet 文件：

1DESCRIBE url('2024.parquet');

1┌──────────────┬──────────────────────┐2│ name         │ type                 │3├──────────────┼──────────────────────┤4│ Id           │ Int64                │5│ PostId       │ Int64                │6│ VoteTypeId   │ Int64                │7│ CreationDate │ DateTime64(3, 'UTC') │8│ UserId       │ Int64                │9│ BountyAmount │ UInt64               │10└──────────────┴──────────────────────┘11126 rows inset. Elapsed: 0.445 sec.

在 MySQL 中创建表

现在让我们连接到 MySQL：

1docker exec -it parquet-mysql mysql -u admin -p stackoverflow

然后运行以下查询来创建一个表来存储这些数据：

1CREATE TABLE votes_from_parquet2(3    Id           BIGINTNOT NULLPRIMARY KEY,4    PostId       BIGINT,5    VoteTypeId   BIGINTNOT NULL,6    CreationDate DATETIME(3) NOT NULL,7    UserId       BIGINT,8    BountyAmount BIGINT UNSIGNED DEFAULT09)10ENGINE = InnoDB;

1Query OK, 0 rows affected (0.039 sec)

在 MySQL 中插入数据

回到 ClickHouse 来摄取数据。我们将使用 mysql 表函数，如下面的查询所示：

1INSERT INTOFUNCTION mysql(2    {mysql_host:String},     -- MySQL host3'stackoverflow',         -- MySQL database4'votes_from_parquet',    -- MySQL table5'admin',                 -- MySQL user6    {mysql_password:String}  -- MySQL password7)8(Id, PostId, VoteTypeId, CreationDate, UserId, BountyAmount)9SELECT Id, PostId, VoteTypeId, CreationDate, UserId, BountyAmount10FROM url('2024.parquet');

12296977 rows inset. Elapsed: 16.609 sec. Processed 2.30 million rows, 21.63 MB (138.29 thousand rows/s., 1.30 MB/s.)2Peak memory usage: 227.31 MiB.

从 ClickHouse 查询 MySQL

我们还可以通过在 mysql 子句中使用 FROM 表函数来查询 MySQL：

1SELECT*2FROM mysql(3    {mysql_host:String},4'stackoverflow',5    (6SELECTcount(*) AS `votes`7FROM votes_from_parquet8WHERE tuple(VoteTypeId, PostId) IN ((2, 37996101), (2, 802038))9    ),10'admin',11    {mysql_password:String}12);

当我们像这样传入内部查询时，它使用 ClickHouse 语法。ClickHouse 将其解析为语法树，将支持的 ClickHouse 构造转换为 MySQL 有效的 SQL，然后将生成的查询发送到 MySQL 执行。

1┏━━━━━━━┓2┃ votes ┃3┡━━━━━━━┩4│     5 │5└───────┘671 row inset. Elapsed: 1.708 sec.

或者，我们可以将查询作为字符串传递给 query 函数，但这次查询必须是 MySQL-SQL。ClickHouse 将其视为不透明字符串，直接发送到 MySQL，不解析或重写其中的 SQL。如果我们按原样传递它，我们会得到一个异常：

1SELECT*2FROM mysql(3    {mysql_host:String},4'stackoverflow',5    query($sql$6SELECTcount(*) AS `votes`7FROM votes_from_parquet8WHERE tuple(VoteTypeId, PostId) IN ((2, 37996101), (2, 802038))9    $sql$),10'admin',11    {mysql_password:String}12);

1Received exception:2Code: 1000. DB::Exception: mysqlxx::BadQuery: FUNCTION stackoverflow.tuple does not exist while executing query: 'SELECT * FROM (3        SELECT count(*) AS `votes`4        FROM votes_from_parquet5        WHERE tuple(VoteTypeId, PostId) IN ((2, 37996101), (2, 802038))6    ) AS __subquery LIMIT 0' (127.0.0.1:3306). (POCO_EXCEPTION)

tuple 在 MySQL 中不存在，因此它无法解析该查询。但如果我们移除 tuple 并发送此查询，它就会正常工作：

1SELECT*2FROM mysql(3    {mysql_host:String},4'stackoverflow',5    query($sql$6SELECTcount(*) AS `votes`7FROM votes_from_parquet8WHERE (VoteTypeId, PostId) IN ((2, 37996101), (2, 802038))9    $sql$),10'admin',11    {mysql_password:String}12);

使用命名集合从 ClickHouse 查询 MySQL

我们还可以使用在配置文件的命名集合中定义的凭据来查询 MySQL。我们可以用这个查询查询服务器的命名集合：

1SELECT*2FROM system.named_collections3FORMAT Vertical;

1Row 1:2──────3name:         mysql_demo4collection:   {'database':'[HIDDEN]','host':'[HIDDEN]','password':'[HIDDEN]','port':'[HIDDEN]','user':'[HIDDEN]'}5source:       CONFIG6create_query:781 row inset. Elapsed: 0.001 sec.

然后我们可以传入 mysql_demo 作为第一个参数，查询以字符串形式提供：

1SELECT*2FROM mysql(3    mysql_demo,4    query=$sql$5SELECTcount(*)6FROM votes_from_parquet7WHERE (VoteTypeId =2AND CreationDate ='2024-01-01')8    $sql$9);

1┌──────────┐2│ count(*) │3├──────────┤4│ 10067    │5└──────────┘671 row in set. Elapsed: 1.100 sec.

让我们用一个查询来结束，该查询查找同时有点赞和点踩的帖子：

1SELECT*FROM mysql(mysql_demo, query ='2  SELECT PostId, upvotes, downvotes,3         upvotes + downvotes AS total_votes,4         ROUND(5           100.0 * LEAST(upvotes, downvotes) / (upvotes + downvotes),6           17         ) AS balance_pct8        FROM9        (10          SELECT PostId, SUM(VoteTypeId = 2) AS upvotes, 11                 SUM(VoteTypeId = 3) AS downvotes12          FROM votes_from_parquet13          WHERE VoteTypeId IN (2, 3)14          GROUP BY PostId15        ) AS vote_totals16  WHERE upvotes > 0 AND downvotes > 017  ORDER BY LEAST(upvotes, downvotes) DESC, total_votes DESC18  LIMIT 10'19);

1┌──────────┬─────────┬───────────┬─────────────┬─────────────┐2│ PostId   │ upvotes │ downvotes │ total_votes │ balance_pct │3├──────────┼─────────┼───────────┼─────────────┼─────────────┤4│ 69699772 │ 205     │ 23        │ 228         │ 10.1        │5│ 69713899 │ 53      │ 9         │ 62          │ 14.5        │6│ 62099904 │ 37      │ 8         │ 45          │ 17.8        │7│ 10032024 │ 7       │ 16        │ 23          │ 30.4        │8│ 78114790 │ 7       │ 12        │ 19          │ 36.8        │9│ 62477194 │ 7       │ 9         │ 16          │ 43.8        │10│ 72231704 │ 8       │ 7         │ 15          │ 46.7        │11│ 77766725 │ 7       │ 8         │ 15          │ 46.7        │12│ 78129981 │ 24      │ 6         │ 30          │ 20          │13│ 77767692 │ 6       │ 8         │ 14          │ 42.9        │14└──────────┴─────────┴───────────┴─────────────┴─────────────┘151610 rows inset. Elapsed: 8.883 sec.

结论

在这篇博客文章中，我们使用 ClickHouse 读取了一个 Parquet 文件，将其内容加载到 MySQL 中，并查询了结果。

能够在不离开 ClickHouse 的情况下完成所有这些操作，这就是我如此喜欢表函数的原因。而且 MySQL 只是其中一个选项。还有用于 Postgres、SQLite 和 MongoDB 的表函数，仅举几例。