Kafka 커넥터 튜토리얼
Kafka 커넥터 튜토리얼 (Kafka connector tutorial)
이 튜토리얼은 Trino에서 Apache Kafka의 라이브 토픽 데이터에 접근하는 방법과, Trino 테이블을 뒷받침하는 토픽 설명 파일을 만드는 방법을 보여줘요.
출처: 문서
본문
소개 (Introduction)
Trino용 Kafka 커넥터를 사용하면 Trino로 Apache Kafka의 라이브 토픽 데이터에 접근할 수 있어요. 이 튜토리얼은 토픽을 설정하는 방법과 Trino 테이블을 뒷받침하는 토픽 설명 파일을 만드는 방법을 보여줘요.
설치 (Installation)
이 튜토리얼은 Trino에 익숙하고 로컬 Trino 설치가 동작한다고 가정해요(Trino 배포 참고). Apache Kafka 설정과 Trino 통합에 초점을 맞출게요.
1단계: Apache Kafka 설치 (Step 1: Install Apache Kafka)
Apache Kafka를 다운로드하고 압축을 풀어요.
참고 (Note)
이 튜토리얼은 Apache Kafka 0.8.1로 테스트했어요. Apache Kafka의 0.8.x 버전에서 동작해야 해요.
ZooKeeper와 Kafka 서버를 시작해요:
$ bin/zookeeper-server-start.sh config/zookeeper.properties
[2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
...
$ bin/kafka-server-start.sh config/server.properties
[2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)
[2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)
...
2단계: TPCH 데이터 로드 (Step 2: Load TPCH data)
TPCH 데이터를 Kafka에 로드하는 도구(kafka-tpch)를 받아요:
$ curl -o kafka-tpch https://repo1.maven.org/maven2/de/softwareforge/kafka_tpch_0811/1.0/kafka_tpch_0811-1.0.sh
$ chmod 755 kafka-tpch
TPCH(tiny) 데이터를 tpch. 프리픽스의 토픽으로 로드해요:
$ ./kafka-tpch load --brokers localhost:9092 --prefix tpch. --tpch-type tiny
2014-07-28T17:17:07.594-0700 INFO main io.airlift.log.Logging Logging to stderr
2014-07-28T17:17:07.623-0700 INFO main de.softwareforge.kafka.LoadCommand Processing tables: [customer, orders, lineitem, part, partsupp, supplier, nation, region]
2014-07-28T17:17:07.981-0700 INFO pool-1-thread-1 de.softwareforge.kafka.LoadCommand Loading table 'customer' into topic 'tpch.customer'...
...
2014-07-28T17:17:13.877-0700 INFO pool-1-thread-3 de.softwareforge.kafka.LoadCommand Generated 60175 rows for table 'lineitem'.
로드 도중 일부 "Failed to fetch topic metadata" 오류는 새 토픽 생성 시점의 정상적인 지연 때문에 나타나며, 이후 재시도로 처리돼요.
3단계: Kafka 커넥터 구성 (Step 3: Configure the Kafka connector)
Kafka 커넥터에서 TPCH 테이블을 노출하려면 카탈로그 속성 파일에 kafka.table-names을 지정해요. 내부 컬럼을 보이게 하는 kafka.hide-internal-columns=false도 설정해요:
connector.name=kafka
kafka.nodes=localhost:9092
kafka.table-names=tpch.customer,tpch.orders,tpch.lineitem,tpch.part,tpch.partsupp,tpch.supplier,tpch.nation,tpch.region
kafka.hide-internal-columns=false
Trino를 시작하고:
$ bin/launcher start
CLI로 접속해 쿼리합니다:
$ ./trino --catalog kafka --schema tpch
trino:tpch> SHOW TABLES;
Table
----------
customer
lineitem
nation
orders
part
partsupp
region
supplier
(8 rows)
토픽 설명 파일 없이는 각 테이블은 내부 컬럼(파티션, 오프셋, 키, 메시지)만 노출해요:
trino:tpch> DESCRIBE customer;
Column | Type | Extra | Comment
-------------------+------------+-------+---------------------------------------------
_partition_id | bigint | | Partition Id
_partition_offset | bigint | | Offset for the message within the partition
_key | varchar | | Key text
_key_corrupt | boolean | | Key data is corrupt
_key_length | bigint | | Total number of key bytes
_message | varchar | | Message text
_message_corrupt | boolean | | Message data is corrupt
_message_length | bigint | | Total number of message bytes
_timestamp | timestamp | | Message timestamp
(11 rows)
trino:tpch> SELECT count(*) FROM customer;
_col0
-------
1500
trino:tpch> SELECT _message FROM customer LIMIT 5;
...
{"rowNumber":1,"customerKey":1,"name":"Customer#000000001","address":"IVhzIApeRb ot,c,E","nationKey":15,"phone":"25-989-741-2988","accountBalance":711.56,"marketSegment":"BUILDING","comment":"to the even, regular platelets. regular, ironic epitaphs nag e"}
...
(5 rows)
trino:tpch> SELECT sum(cast(json_extract_scalar(_message, '$.accountBalance') AS DOUBLE)) FROM customer LIMIT 10;
_col0
------------
6681865.59
(1 row)
4단계: 토픽 설명 파일로 스키마 부여 (Step 4: Add a topic description file)
내부 컬럼 대신 실제 스키마 컬럼을 만들려면 토픽 설명 파일을 작성해요. Kafka 커넥터는 etc/kafka/ 디렉토리에서 테이블명과 같은 이름의 JSON 파일을 읽어요. 먼저 키 컬럼만 추가하는 예시예요:
{
"tableName": "customer",
"schemaName": "tpch",
"topicName": "tpch.customer",
"key": {
"dataFormat": "raw",
"fields": [
{
"name": "kafka_key",
"dataFormat": "LONG",
"type": "BIGINT",
"hidden": "false"
}
]
}
}
이제 kafka_key 컬럼이 보여요:
trino:tpch> DESCRIBE customer;
Column | Type | Extra | Comment
-------------------+------------+-------+---------------------------------------------
kafka_key | bigint | |
_partition_id | bigint | | Partition Id
...
(12 rows)
trino:tpch> SELECT kafka_key FROM customer ORDER BY kafka_key LIMIT 10;
kafka_key
-----------
0
1
2
3
4
5
6
7
8
9
(10 rows)
메시지를 JSON으로 파싱해 실제 데이터 컬럼을 추가할 수 있어요:
{
"tableName": "customer",
"schemaName": "tpch",
"topicName": "tpch.customer",
"key": {
"dataFormat": "raw",
"fields": [
{
"name": "kafka_key",
"dataFormat": "LONG",
"type": "BIGINT",
"hidden": "false"
}
]
},
"message": {
"dataFormat": "json",
"fields": [
{
"name": "row_number",
"mapping": "rowNumber",
"type": "BIGINT"
},
{
"name": "customer_key",
"mapping": "customerKey",
"type": "BIGINT"
},
{
"name": "name",
"mapping": "name",
"type": "VARCHAR"
},
{
"name": "address",
"mapping": "address",
"type": "VARCHAR"
},
{
"name": "nation_key",
"mapping": "nationKey",
"type": "BIGINT"
},
{
"name": "phone",
"mapping": "phone",
"type": "VARCHAR"
},
{
"name": "account_balance",
"mapping": "accountBalance",
"type": "DOUBLE"
},
{
"name": "market_segment",
"mapping": "marketSegment",
"type": "VARCHAR"
},
{
"name": "comment",
"mapping": "comment",
"type": "VARCHAR"
}
]
}
}
이제 실제 스키마 컬럼들로 쿼리할 수 있어요:
trino:tpch> DESCRIBE customer;
Column | Type | Extra | Comment
-------------------+------------+-------+---------------------------------------------
kafka_key | bigint | |
row_number | bigint | |
customer_key | bigint | |
name | varchar | |
address | varchar | |
nation_key | bigint | |
phone | varchar | |
account_balance | double | |
market_segment | varchar | |
comment | varchar | |
_partition_id | bigint | | Partition Id
...
(21 rows)
trino:tpch> SELECT * FROM customer LIMIT 5;
kafka_key | row_number | customer_key | name | ...
----------+------------+--------------+--------------------+...
1 | 2 | 2 | Customer#000000002 | ...
...
(5 rows)
trino:tpch> SELECT sum(account_balance) FROM customer LIMIT 10;
_col0
------------
6681865.59
(1 row)
5단계: 라이브 데이터 — Twitter 예시 (Step 5: Live data)
라이브 스트림을 소비하는 예시로 Twitter 피드를 Kafka로 보내는 도구(twistr)를 쓸 수 있어요:
$ curl -o twistr https://repo1.maven.org/maven2/de/softwareforge/twistr_kafka_0811/1.2/twistr_kafka_0811-1.2.sh
$ chmod 755 twistr
twistr.properties 설정:
twistr.access-token-key=...
twistr.access-token-secret=...
twistr.consumer-key=...
twistr.consumer-secret=...
twistr.kafka.brokers=localhost:9092
토픽 설명 파일 tweets.json(예시 축약), JSON 메시지의 중첩 필드는 슬래시 경로(user/screen_name)로 매핑하고, 날짜는 rfc2822 포맷으로 파싱해요:
{
"tableName": "tweets",
"topicName": "twitter_feed",
"dataFormat": "json",
"key": {
"dataFormat": "raw",
"fields": [
{
"name": "kafka_key",
"dataFormat": "LONG",
"type": "BIGINT",
"hidden": "false"
}
]
},
"message": {
"dataFormat":"json",
"fields": [
{
"name": "text",
"mapping": "text",
"type": "VARCHAR"
},
{
"name": "user_name",
"mapping": "user/screen_name",
"type": "VARCHAR"
},
{
"name": "lang",
"mapping": "lang",
"type": "VARCHAR"
},
{
"name": "created_at",
"mapping": "created_at",
"type": "TIMESTAMP",
"dataFormat": "rfc2822"
},
{
"name": "favorite_count",
"mapping": "favorite_count",
"type": "BIGINT"
},
{
"name": "retweet_count",
"mapping": "retweet_count",
"type": "BIGINT"
},
{
"name": "favorited",
"mapping": "favorited",
"type": "BOOLEAN"
},
{
"name": "id",
"mapping": "id_str",
"type": "VARCHAR"
},
{
"name": "in_reply_to_screen_name",
"mapping": "in_reply_to_screen_name",
"type": "VARCHAR"
},
{
"name": "place_name",
"mapping": "place/full_name",
"type": "VARCHAR"
}
]
}
}
twistr를 실행하고:
$ java -Dness.config.location=file:$(pwd) -Dness.config=twistr -jar ./twistr
Trino에서 라이브 토픽을 조회해요. 데이터가 계속 흘러들어오므로 count가 매번 달라져요:
$ ./trino --catalog kafka --schema default
trino:default> SELECT count(*) FROM tweets;
_col0
-------
4467
(1 row)
trino:default> SELECT count(*) FROM tweets;
_col0
-------
4517
(1 row)
trino:default> SELECT kafka_key, user_name, lang, created_at FROM tweets LIMIT 10;
kafka_key | user_name | lang | created_at
--------------------+-----------------+------+-------------------------
494227746231685121 | burncaniff | en | 2014-07-29 14:07:31.000
...
(10 rows)
원시 날짜 문자열을 확인해 보면:
trino:default> SELECT DISTINCT json_extract_scalar(_message, '$.created_at')) AS raw_date
-> FROM tweets LIMIT 5;
raw_date
--------------------------------
Tue Jul 29 21:07:31 +0000 2014
...
(5 rows)
토픽 설명의 dataFormat: rfc2822 덕분에 created_at 컬럼이 올바른 TIMESTAMP로 파싱된 것을 확인할 수 있어요:
trino:default> SELECT created_at, raw_date FROM (
-> SELECT created_at, json_extract_scalar(_message, '$.created_at') AS raw_date
-> FROM tweets)
-> GROUP BY 1, 2 LIMIT 5;
created_at | raw_date
-------------------------+--------------------------------
2014-07-29 14:07:20.000 | Tue Jul 29 21:07:20 +0000 2014
...
(5 rows)
더 알아보기 (Learn more)
토픽 설명 파일의 포맷과 옵션이 궁금하다면 Kafka 커넥터 문서를 이어서 읽어 보세요.