> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-mintlify-55d9d317.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 如何配合 Kafka 使用新的 JSON 数据类型？

> 了解如何使用 Kafka 表引擎和 JSON 数据类型，将 Apache Kafka 中的 JSON 消息直接加载到 ClickHouse 的单个 JSON 列中。

{frontMatter.description}

<div id="kafka-and-the-json-data-type">
  ## Kafka 与 JSON 数据类型
</div>

随着新的 [`JSON`](/zh/reference/data-types/newjson) 数据类型推出，ClickHouse 现已成为进行 [JSON 分析](https://clickhouse.com/engineering-resources/json-database) 的理想数据库选择。
在本指南中，我们将学习如何将 Apache Kafka 中的 JSON 消息直接加载到 ClickHouse 的单个 `JSON` 列中。

<div id="setup-kafka">
  ## 设置 Kafka
</div>

首先，在我们的机器上运行一个 Kafka broker。我们还会将 9092 端口映射到主机操作系统的 9092 端口，以便更方便地与 Kafka 交互：

```bash theme={null}
docker run --name broker -p 9092:9092 apache/kafka:3.8.1
```

<div id="ingest-data-into-kafka">
  ## 将数据摄取到 Kafka
</div>

运行起来后，我们需要摄取一些数据。
Wikimedia 的 recent changes feed 是一个很好的流式数据源，因此我们将其摄取到 `wiki_events` topic 中：

```bash theme={null}
curl -N https://stream.wikimedia.org/v2/stream/recentchange 2>/dev/null |
awk '/^data: /{gsub(/^data: /, ""); print}' |
jq -cr --arg sep ø '[.meta.id, tostring] | join($sep)' |
kcat -P -b localhost:9092 -t wiki_events -Kø
```

我们可以运行以下命令，检查数据是否正在被摄取：

```bash theme={null}
kcat -C -b localhost:9092  -t wiki_events
```

```text theme={null}
{"$schema":"/mediawiki/recentchange/1.0.0","meta":{"uri":"https://www.wikidata.org/wiki/Q130972321","request_id":"5c687ded-4721-4bfc-ae6c-58ca25f4a6ce","id":"0fbb0982-c43b-4e8b-989b-db7e78dbdc76","dt":"2024-11-06T11:59:57Z","domain":"www.wikidata.org","stream":"mediawiki.recentchange","topic":"codfw.mediawiki.recentchange","partition":0,"offset":1228777205},"id":2338656448,"type":"edit","namespace":0,"title":"Q130972321","title_url":"https://www.wikidata.org/wiki/Q130972321","comment":"/* wbsetclaim-create:2||1 */ [[Property:P18]]: Mahdi Rrezaei Journalist.jpg","timestamp":1730894397,"user":"Wikimellatir","bot":false,"notify_url":"https://www.wikidata.org/w/index.php?diff=2270885254&oldid=2270870214&rcid=2338656448","minor":false,"patrolled":false,"length":{"old":4269,"new":4636},"revision":{"old":2270870214,"new":2270885254},"server_url":"https://www.wikidata.org","server_name":"www.wikidata.org","server_script_path":"/w","wiki":"wikidatawiki","parsedcomment":"<span dir=\"auto\"><span class=\"autocomment\">Created claim: </span></span> <a href=\"/wiki/Property:P18\" title=\"image | image of relevant illustration of the subject; if available, also use more specific properties (sample: coat of arms image, locator map, flag image, signature image, logo image, collage image)\"><span class=\"wb-itemlink\"><span class=\"wb-itemlink-label\" lang=\"en\" dir=\"ltr\">image</span> <span class=\"wb-itemlink-id\">(P18)</span></span></a>: Mahdi Rrezaei Journalist.jpg"}
{"$schema":"/mediawiki/recentchange/1.0.0","meta":{"uri":"https://www.wikidata.org/wiki/Q75756596","request_id":"eb116219-7372-4725-986f-790211708d36","id":"9e0d5299-5bd1-4c58-b796-9852afd8a84e","dt":"2024-11-06T11:59:54Z","domain":"www.wikidata.org","stream":"mediawiki.recentchange","topic":"codfw.mediawiki.recentchange","partition":0,"offset":1228777206},"id":2338656449,"type":"edit","namespace":0,"title":"Q75756596","title_url":"https://www.wikidata.org/wiki/Q75756596","comment":"/* wbeditentity-update-languages-and-other:0||55 */ mv labels and aliases matching [[Property:P528]] or [[Property:P3083]] to mul","timestamp":1730894394,"user":"Twofivesixbot","bot":true,"notify_url":"https://www.wikidata.org/w/index.php?diff=2270885237&oldid=2147709089&rcid=2338656449","minor":false,"patrolled":true,"length":{"old":30879,"new":27161},"revision":{"old":2147709089,"new":2270885237},"server_url":"https://www.wikidata.org","server_name":"www.wikidata.org","server_script_path":"/w","wiki":"wikidatawiki","parsedcomment":"<span dir=\"auto\"><span class=\"autocomment\">Changed label, description and/or aliases in 55 languages, and other parts: </span></span> mv labels and aliases matching <a href=\"/wiki/Property:P528\" title=\"catalog code | catalog name of an object, use with qualifier P972\"><span class=\"wb-itemlink\"><span class=\"wb-itemlink-label\" lang=\"en\" dir=\"ltr\">catalog code</span> <span class=\"wb-itemlink-id\">(P528)</span></span></a> or <a href=\"/wiki/Property:P3083\" title=\"SIMBAD ID | identifier for an astronomical object, in the University of Strasbourg&#039;s SIMBAD database\"><span class=\"wb-itemlink\"><span class=\"wb-itemlink-label\" lang=\"en\" dir=\"ltr\">SIMBAD ID</span> <span class=\"wb-itemlink-id\">(P3083)</span></span></a> to mul"}
```

到这里一切顺利。

<div id="ingest-data-into-clickhouse">
  ## 将数据摄取到 ClickHouse
</div>

接下来，我们将把数据摄取到 ClickHouse 中。
首先，通过设置以下属性来启用 JSON 类型 (目前仍处于 Experimental 阶段) ：

```sql theme={null}
SET allow_experimental_json_type = 1;
```

现在，我们来创建 `wiki_queue` 表，它使用 [`Kafka` 表引擎](/zh/integrations/connectors/data-ingestion/kafka/kafka-table-engine)。

```sql theme={null}
CREATE TABLE wiki_queue
(
    json JSON
)
ENGINE = Kafka(
  'localhost:9092', 
  'wiki_events', 
  'clickhouse-consumer-group',
  'JSONAsObject'
);
```

请注意，这里我们使用的是 [`JSONAsObject`](/zh/reference/formats/JSON/JSONAsObject) 格式，以确保传入的消息可作为 JSON 对象使用。
这种格式只能解析到仅有一个 `JSON` 类型列的表中。

接下来，我们将创建用于存储 Wiki 数据的底层表：

```sql theme={null}
CREATE TABLE wiki
(
    json JSON,
    id String MATERIALIZED getSubcolumn(json, 'meta.id')
)
ENGINE = MergeTree
ORDER BY id;
```

最后，创建一个 materialized view 来向 `wiki` 表填充数据：

```sql theme={null}
CREATE MATERIALIZED VIEW wiki_mv TO wiki AS 
SELECT json
FROM wiki_queue;
```

<div id="querying-json-data-in-clickhouse">
  ## 在 ClickHouse 中查询 JSON 数据
</div>

然后，我们就可以查询 `wiki` 表了。
例如，我们可以统计提交过更改的机器人数量：

```sql theme={null}
SELECT json.bot, count()
FROM wiki
GROUP BY ALL
```

```text theme={null}
   ┌─json.bot─┬─count()─┐
1. │ true     │    2526 │
2. │ false    │    4691 │
   └──────────┴─────────┘
```

或者，我们也可以找出在 `en.wikipedia.org` 上编辑次数最多的用户：

```sql theme={null}
SELECT
    json.user,
    count()
FROM wiki
WHERE json.server_name = 'en.wikipedia.org'
GROUP BY ALL
ORDER BY count() DESC
LIMIT 10
```

```text theme={null}
    ┌─json.user──────────────────────────────┬─count()─┐
 1. │ Monkbot                                │     267 │
 2. │ Onel5969                               │     107 │
 3. │ Bangwiki                               │      37 │
 4. │ HHH Pedrigree                          │      28 │
 5. │ REDACTED403                            │      23 │
 6. │ KylieTastic                            │      22 │
 7. │ Tinniesbison                           │      21 │
 8. │ XTheBedrockX                           │      20 │
 9. │ 2001:4455:1DB:4000:51F3:6A16:408E:69FC │      19 │
10. │ Wcquidditch                            │      15 │
    └────────────────────────────────────────┴─────────┘
```