> ## Documentation Index > Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt > Use this file to discover all available pages before exploring further. # KafkaからClickHouseにデータを取り込む方法 > KafkaトピックからClickHouseへデータを取り込む方法を、Kafkaテーブルエンジン、materialized view、MergeTreeテーブルを使って学びます。 {frontMatter.description}
## 概要
この記事では、Kafka トピックから ClickHouse テーブルへデータを送信する手順を説明します。使用するのは Wiki の recent changes フィードで、これはさまざまな Wikimedia プロパティに加えられた変更を表す[イベントストリーム](https://stream.wikimedia.org/v2/stream/recentchange)を提供します。手順は次のとおりです。 1. Ubuntu で Kafka をセットアップする方法 2. データストリームを Kafka トピックに取り込む 3. トピックを購読する ClickHouse テーブルを作成する
# 1. Ubuntu で Kafka をセットアップ
1. Ubuntu **ec2** インスタンスを作成し、SSH で接続します: ```bash theme={null} ssh -i ~/training.pem ubuntu@ec2.compute.amazonaws.com ``` 2. Kafka をインストールします (以下の手順を参照: [https://www.linode.com/docs/guides/how-to-install-apache-kafka-on-ubuntu/](https://www.linode.com/docs/guides/how-to-install-apache-kafka-on-ubuntu/)) : ```bash theme={null} sudo apt update sudo apt install openjdk-11-jdk mkdir /home/ubuntu/kafka cd /home/ubuntu/kafka/ wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz tar -zxvf kafka_2.13-3.7.0.tgz ``` 3. ZooKeeper を起動します。 ```bash theme={null} cd kafka_2.13-3.7.0 bin/zookeeper-server-start.sh config/zookeeper.properties ``` 4. 新しいターミナルを開いて、Kafka を起動します: ```bash theme={null} ssh -i ~/training.pem ubuntu@ec2.compute.amazonaws.com cd kafka/kafka_2.13-3.7.0/ bin/kafka-server-start.sh config/server.properties ``` 5. 3つ目のターミナルを開き、`wikimedia` という名前のトピックを作成します: ```bash theme={null} ssh -i ~/training.pem ubuntu@ec2.compute.amazonaws.com cd kafka/kafka_2.13-3.7.0/ bin/kafka-topics.sh --create --topic wikimedia --bootstrap-server localhost:9092 ``` 6. 正常に作成されたことを、次の方法で確認できます： ```bash theme={null} bin/kafka-topics.sh --list --bootstrap-server localhost:9092 ```
# 2. Wikimedia Stream を Kafka に取り込む
1. まず、いくつかのユーティリティが必要です。 ```bash theme={null} sudo apt-get install librdkafka-dev libyajl-dev sudo apt-get install kafkacat ``` 2. データは、最新の Wikimedia イベントを取得し、JSON データを抽出して Kafka トピックに送信する、巧妙な **curl** コマンドを使って Kafka に送られます。 ```bash theme={null} curl -N https://stream.wikimedia.org/v2/stream/recentchange | awk '/^data: /{gsub(/^data: /, ""); print}' | kafkacat -P -b localhost:9092 -t wikimedia ``` 3. トピックを"describe"して確認できます: ```bash theme={null} bin/kafka-topics.sh --describe --topic wikimedia --bootstrap-server localhost:9092 ``` 4. いくつかのイベントを取り込んで、すべてが正常に動作していることを確認しましょう。 ```bash theme={null} bin/kafka-console-consumer.sh --topic wikimedia --from-beginning --bootstrap-server localhost:9092 ``` 5. 前のコマンドを停止するには、**Ctrl+c** を押します。
# 3. データを ClickHouse に取り込む
1. 取り込まれるデータは以下のようになります。 ```json theme={null} { "$schema": "/mediawiki/recentchange/1.0.0", "meta": { "uri": "https://www.wikidata.org/wiki/Q45791749", "request_id": "f64cfb17-04ba-4d09-8935-38ec6f0001c2", "id": "9d7d2b5a-b79b-45ea-b72c-69c3b69ae931", "dt": "2024-04-18T13:21:21Z", "domain": "www.wikidata.org", "stream": "mediawiki.recentchange", "topic": "eqiad.mediawiki.recentchange", "partition": 0, "offset": 5032636513 }, "id": 2196113017, "type": "edit", "namespace": 0, "title": "Q45791749", "title_url": "https://www.wikidata.org/wiki/Q45791749", "comment": "/* wbsetqualifier-add:1| */ [[Property:P1545]]: 20, Modify PubMed ID: 7292984 citation data from NCBI, Europe PMC and CrossRef", "timestamp": 1713446481, "user": "Cewbot", "bot": true, "notify_url": "https://www.wikidata.org/w/index.php?diff=2131981357&oldid=2131981341&rcid=2196113017", "minor": false, "patrolled": true, "length": { "old": 75618, "new": 75896 }, "revision": { "old": 2131981341, "new": 2131981357 }, "server_url": "https://www.wikidata.org", "server_name": "www.wikidata.org", "server_script_path": "/w", "wiki": "wikidatawiki", "parsedcomment": "Added qualifier: series ordinal (P1545): 20, Modify PubMed ID: 7292984 citation data from NCBI, Europe PMC and CrossRef" } ``` 2. **Kafka** テーブルエンジンを使用して Kafkaトピックからデータを取り込む必要があります： ```sql theme={null} CREATE OR REPLACE TABLE wikiQueue ( `id` UInt32, `type` String, `title` String, `title_url` String, `comment` String, `timestamp` UInt64, `user` String, `bot` Bool, `server_url` String, `server_name` String, `wiki` String, `meta` Tuple(uri String, id String, stream String, topic String, domain String) ) ENGINE = Kafka( 'ec2.compute.amazonaws.com:9092', 'wikimedia', 'consumer-group-wiki', 'JSONEachRow' ); ``` 3. 何らかの理由で、**Kafka**の**テーブルエンジン**がパブリックな **ec2** URL をプライベート DNS 名に変換してしまうようなので、ローカルの`/etc/hosts`ファイルにもそれを追加する必要がありました。 ```bash theme={null} 52.14.154.92 ip.us-east-2.compute.internal ``` 4. Kafkaテーブルから読み取るには、設定を有効にするだけです。 ```sql theme={null} SELECT * FROM wikiQueue LIMIT 20 FORMAT Vertical SETTINGS stream_like_engine_allow_direct_select = 1; ``` **wikiQueue** テーブルで定義されたカラムに基づいて、行データは適切にパースされた状態で返されるはずです: ```response theme={null} id: 2473996741 type: edit title: File:Père-Lachaise - Division 6 - Cassereau 05.jpg title_url: https://commons.wikimedia.org/wiki/File:P%C3%A8re-Lachaise_-_Division_6_-_Cassereau_05.jpg comment: /* wbcreateclaim-create:1| */ [[d:Special:EntityPage/P921]]: [[d:Special:EntityPage/Q112327116]], [[:toollabs:quickstatements/#/batch/228454|batch #228454]] timestamp: 1713457283 user: Ameisenigel bot: false server_url: https://commons.wikimedia.org server_name: commons.wikimedia.org wiki: commonswiki meta: ('https://commons.wikimedia.org/wiki/File:P%C3%A8re-Lachaise_-_Division_6_-_Cassereau_05.jpg','01a832e2-24c5-4ccb-bd93-8e2c0e429418','mediawiki.recentchange','eqiad.mediawiki.recentchange','commons.wikimedia.org') ``` 5. これらの受信イベントを保存するには、**MergeTree** テーブルが必要です。 ```sql theme={null} CREATE TABLE rawEvents ( id UInt64, type LowCardinality(String), comment String, timestamp DateTime64(3, 'UTC'), title_url String, topic LowCardinality(String), user String ) ENGINE = MergeTree ORDER BY (type, timestamp); ``` 6. **Kafka** テーブルで insert が行われたときにトリガーされ、データを **rawEvents** テーブルに送る materialized view を定義しましょう: ```sql theme={null} CREATE MATERIALIZED VIEW rawEvents_mv TO rawEvents AS SELECT id, type, comment, toDateTime(timestamp) AS timestamp, title_url, tupleElement(meta, 'topic') AS topic, user FROM wikiQueue WHERE title_url <> ''; ``` 7. ほぼすぐに、データが **rawEvents** に取り込まれ始めるはずです： ```sql theme={null} SELECT count() FROM rawEvents; ``` 8. いくつかの行を確認してみましょう: ```sql theme={null} SELECT * FROM rawEvents LIMIT 5 FORMAT Vertical ``` ```response theme={null} Row 1: ────── id: 124842852 type: 142 comment: Pere prlpz commented on "Plantilles Enciclopèdia Catalana" (Diria que no cal fer res als articles. Es pot actualitzar els enllaços que es facin servir a les referències (tot i que l'antic encara ha...) timestamp: 2024-04-18 16:22:29.000 title_url: https://ca.wikipedia.org/wiki/Tema:Wu36d6vfsiuu4jsi topic: eqiad.mediawiki.recentchange user: Pere prlpz Row 2: ────── id: 2473996748 type: categorize comment: [[:File:Ruïne van een poortgebouw, RP-T-1976-29-6(R).jpg]] removed from category timestamp: 2024-04-18 16:21:20.000 title_url: https://commons.wikimedia.org/wiki/Category:Pieter_Moninckx topic: eqiad.mediawiki.recentchange user: Warburg1866 Row 3: ────── id: 311828596 type: categorize comment: [[:Cujo (película)]] añadida a la categoría timestamp: 2024-04-18 16:21:21.000 title_url: https://es.wikipedia.org/wiki/Categor%C3%ADa:Pel%C3%ADculas_basadas_en_obras_de_Stephen_King topic: eqiad.mediawiki.recentchange user: Beta15 Row 4: ────── id: 311828597 type: categorize comment: [[:Cujo (película)]] eliminada de la categoría timestamp: 2024-04-18 16:21:21.000 title_url: https://es.wikipedia.org/wiki/Categor%C3%ADa:Trabajos_basados_en_obras_de_Stephen_King topic: eqiad.mediawiki.recentchange user: Beta15 Row 5: ────── id: 48494536 type: categorize comment: [[:braiteremmo]] ajoutée à la catégorie timestamp: 2024-04-18 16:21:21.000 title_url: https://fr.wiktionary.org/wiki/Cat%C3%A9gorie:Wiktionnaire:Exemples_manquants_en_italien topic: eqiad.mediawiki.recentchange user: Àncilu bot ``` 9. どのようなイベントが入ってきているか見てみましょう： ``` SELECT type, count() FROM rawEvents GROUP BY type ``` ```response theme={null} ┌─type───────┬─count()─┐ 1. │ 142 │ 1 │ 2. │ new │ 1003 │ 3. │ categorize │ 12228 │ 4. │ log │ 1799 │ 5. │ edit │ 17142 │ └────────────┴─────────┘ ``` 現在の materialized view に連なる materialized view を定義しましょう。分単位の集計統計を追跡します。 ```sql theme={null} CREATE TABLE byMinute ( `dateTime` DateTime64(3, 'UTC') NOT NULL, `users` AggregateFunction(uniq, String), `pages` AggregateFunction(uniq, String), `updates` AggregateFunction(sum, UInt32) ) ENGINE = AggregatingMergeTree ORDER BY dateTime; CREATE MATERIALIZED VIEW byMinute_mv TO byMinute AS SELECT toStartOfMinute(timestamp) AS dateTime, uniqState(user) AS users, uniqState(title_url) AS pages, sumState(toUInt32(1)) AS updates FROM rawEvents GROUP BY dateTime; ``` 9. 結果を確認するには、**-Merge** 関数が必要です： ```sql theme={null} SELECT dateTime AS dateTime, uniqMerge(users) AS users, uniqMerge(pages) AS pages, sumMerge(updates) AS updates FROM byMinute GROUP BY dateTime ORDER BY dateTime DESC LIMIT 10; ```