Kafka 連接器教學¶
簡介¶
Presto 的 Kafka 連接器允許使用 Presto 從 Apache Kafka 存取即時主題資料。本教學示範如何設定主題，以及如何建立支援 Presto 資料表的主題描述檔。
安裝¶
本教學假設您熟悉 Presto 和運作中的本機 Presto 安裝 (請參閱部署 Presto)。本教學將重點放在設定 Apache Kafka 並將其與 Presto 整合。
步驟 1：安裝 Apache Kafka¶
下載並解壓縮Apache Kafka。
注意
本教學已使用 Apache Kafka 0.8.1 測試。它應適用於任何 0.8.x 版本的 Apache Kafka。
啟動 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)
...
這將在連接埠 2181 上啟動 Zookeeper，並在連接埠 9092 上啟動 Kafka。
步驟 2：載入資料¶
從 Maven 中央下載 tpch-kafka 載入器
$ 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
現在執行 kafka-tpch 程式，以使用 tpch 資料預先載入多個主題
$ ./kafka-tpch load --brokers localhost:9092 --prefix tpch. --tpch-type tiny
2014-07-28T17:17:07.594-0700 INFO main com.facebook.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:07.981-0700 INFO pool-1-thread-2 de.softwareforge.kafka.LoadCommand Loading table 'orders' into topic 'tpch.orders'...
2014-07-28T17:17:07.981-0700 INFO pool-1-thread-3 de.softwareforge.kafka.LoadCommand Loading table 'lineitem' into topic 'tpch.lineitem'...
2014-07-28T17:17:07.982-0700 INFO pool-1-thread-4 de.softwareforge.kafka.LoadCommand Loading table 'part' into topic 'tpch.part'...
2014-07-28T17:17:07.982-0700 INFO pool-1-thread-5 de.softwareforge.kafka.LoadCommand Loading table 'partsupp' into topic 'tpch.partsupp'...
2014-07-28T17:17:07.982-0700 INFO pool-1-thread-6 de.softwareforge.kafka.LoadCommand Loading table 'supplier' into topic 'tpch.supplier'...
2014-07-28T17:17:07.982-0700 INFO pool-1-thread-7 de.softwareforge.kafka.LoadCommand Loading table 'nation' into topic 'tpch.nation'...
2014-07-28T17:17:07.982-0700 INFO pool-1-thread-8 de.softwareforge.kafka.LoadCommand Loading table 'region' into topic 'tpch.region'...
2014-07-28T17:17:10.612-0700 ERROR pool-1-thread-8 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.region
2014-07-28T17:17:10.781-0700 INFO pool-1-thread-8 de.softwareforge.kafka.LoadCommand Generated 5 rows for table 'region'.
2014-07-28T17:17:10.797-0700 ERROR pool-1-thread-3 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.lineitem
2014-07-28T17:17:10.932-0700 ERROR pool-1-thread-1 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.customer
2014-07-28T17:17:11.068-0700 ERROR pool-1-thread-2 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.orders
2014-07-28T17:17:11.200-0700 ERROR pool-1-thread-6 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.supplier
2014-07-28T17:17:11.319-0700 INFO pool-1-thread-6 de.softwareforge.kafka.LoadCommand Generated 100 rows for table 'supplier'.
2014-07-28T17:17:11.333-0700 ERROR pool-1-thread-4 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.part
2014-07-28T17:17:11.466-0700 ERROR pool-1-thread-5 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.partsupp
2014-07-28T17:17:11.597-0700 ERROR pool-1-thread-7 kafka.producer.async.DefaultEventHandler Failed to collate messages by topic, partition due to: Failed to fetch topic metadata for topic: tpch.nation
2014-07-28T17:17:11.706-0700 INFO pool-1-thread-7 de.softwareforge.kafka.LoadCommand Generated 25 rows for table 'nation'.
2014-07-28T17:17:12.180-0700 INFO pool-1-thread-1 de.softwareforge.kafka.LoadCommand Generated 1500 rows for table 'customer'.
2014-07-28T17:17:12.251-0700 INFO pool-1-thread-4 de.softwareforge.kafka.LoadCommand Generated 2000 rows for table 'part'.
2014-07-28T17:17:12.905-0700 INFO pool-1-thread-2 de.softwareforge.kafka.LoadCommand Generated 15000 rows for table 'orders'.
2014-07-28T17:17:12.919-0700 INFO pool-1-thread-5 de.softwareforge.kafka.LoadCommand Generated 8000 rows for table 'partsupp'.
2014-07-28T17:17:13.877-0700 INFO pool-1-thread-3 de.softwareforge.kafka.LoadCommand Generated 60175 rows for table 'lineitem'.
Kafka 現在有多個預先載入要查詢資料的主題。
步驟 3：讓 Presto 知道 Kafka 主題¶
在您的 Presto 安裝中，為 Kafka 連接器新增目錄屬性檔案 etc/catalog/kafka.properties。此檔案會列出 Kafka 節點和主題
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
現在啟動 Presto
$ bin/launcher start
由於 Kafka 資料表在組態中都具有 tpch. 前置詞，因此資料表位於 tpch 結構描述中。連接器已裝載到 kafka 目錄中，因為屬性檔案命名為 kafka.properties。
$ ./presto --catalog kafka --schema tpch
列出資料表以驗證運作是否正常
presto:tpch> SHOW TABLES;
Table
----------
customer
lineitem
nation
orders
part
partsupp
region
supplier
(8 rows)
步驟 4：基本資料查詢¶
Kafka 資料是非結構化的，而且沒有描述訊息格式的中繼資料。在沒有進一步組態的情況下，Kafka 連接器可以存取資料並以原始格式對應資料，但除了內建的欄位之外，沒有任何實際欄位
presto: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
(11 rows)
presto:tpch> SELECT count(*) FROM customer;
_col0
-------
1500
presto:tpch> SELECT _message FROM customer LIMIT 5;
_message
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
{"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"}
{"rowNumber":3,"customerKey":3,"name":"Customer#000000003","address":"MG9kdTD2WBHm","nationKey":1,"phone":"11-719-748-3364","accountBalance":7498.12,"marketSegment":"AUTOMOBILE","comment":" deposits eat slyly ironic, even instructions. express foxes detect slyly. blithel
{"rowNumber":5,"customerKey":5,"name":"Customer#000000005","address":"KvpyuHCplrB84WgAiGV6sYpZq7Tj","nationKey":3,"phone":"13-750-942-6364","accountBalance":794.47,"marketSegment":"HOUSEHOLD","comment":"n accounts will have to unwind. foxes cajole accor"}
{"rowNumber":7,"customerKey":7,"name":"Customer#000000007","address":"TcGe5gaZNgVePxU5kRrvXBfkasDTea","nationKey":18,"phone":"28-190-982-9759","accountBalance":9561.95,"marketSegment":"AUTOMOBILE","comment":"ainst the ironic, express theodolites. express, even pinto bean
{"rowNumber":9,"customerKey":9,"name":"Customer#000000009","address":"xKiAFTjUsCuxfeleNqefumTrjS","nationKey":8,"phone":"18-338-906-3675","accountBalance":8324.07,"marketSegment":"FURNITURE","comment":"r theodolites according to the requests wake thinly excuses: pending
(5 rows)
presto:tpch> SELECT sum(cast(json_extract_scalar(_message, '$.accountBalance') AS double)) FROM customer LIMIT 10;
_col0
------------
6681865.59
(1 row)
可以使用 Presto 查詢 Kafka 中的資料，但它尚未採用實際的資料表格式。原始資料可透過 _message 和 _key 欄位取得，但尚未解碼為欄位。由於範例資料採用 JSON 格式，因此可以使用 Presto 內建的JSON 函式和運算子來分割資料。
步驟 5：新增主題描述檔¶
Kafka 連接器支援主題描述檔，可將原始資料轉換為資料表格式。這些檔案位於 Presto 安裝中的 etc/kafka 資料夾中，而且必須以 .json 結尾。建議檔案名稱與資料表名稱相符，但這並非必要。
將下列檔案新增為 etc/kafka/tpch.customer.json 並重新啟動 Presto
{
"tableName": "customer",
"schemaName": "tpch",
"topicName": "tpch.customer",
"key": {
"dataFormat": "raw",
"fields": [
{
"name": "kafka_key",
"dataFormat": "LONG",
"type": "BIGINT",
"hidden": "false"
}
]
}
}
customer 資料表現在有一個額外的欄位：kafka_key。
presto:tpch> DESCRIBE customer;
Column | Type | Extra | Comment
-------------------+---------+-------+---------------------------------------------
kafka_key | bigint | |
_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
(12 rows)
presto: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)
主題定義檔案會將內部 Kafka 金鑰 (原始長度為八個位元組) 對應到 Presto BIGINT 欄位。
步驟 6：將主題訊息中的所有值對應到欄位¶
更新 etc/kafka/tpch.customer.json 檔案以新增訊息欄位並重新啟動 Presto。由於訊息中的欄位是 JSON，因此會使用 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"
}
]
}
}
現在針對訊息的 JSON 中的所有欄位，定義欄位，而且稍早的 sum 查詢可以直接在 account_balance 欄位上操作
presto: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
_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
(21 rows)
presto:tpch> SELECT * FROM customer LIMIT 5;
kafka_key | row_number | customer_key | name | address | nation_key | phone | account_balance | market_segment | comment
-----------+------------+--------------+--------------------+---------------------------------------+------------+-----------------+-----------------+----------------+---------------------------------------------------------------------------------------------------------
1 | 2 | 2 | Customer#000000002 | XSTf4,NCwDVaWNe6tEgvwfmRchLXak | 13 | 23-768-687-3665 | 121.65 | AUTOMOBILE | l accounts. blithely ironic theodolites integrate boldly: caref
3 | 4 | 4 | Customer#000000004 | XxVSJsLAGtn | 4 | 14-128-190-5944 | 2866.83 | MACHINERY | requests. final, regular ideas sleep final accou
5 | 6 | 6 | Customer#000000006 | sKZz0CsnMD7mp4Xd0YrBvx,LREYKUWAh yVn | 20 | 30-114-968-4951 | 7638.57 | AUTOMOBILE | tions. even deposits boost according to the slyly bold packages. final accounts cajole requests. furious
7 | 8 | 8 | Customer#000000008 | I0B10bB0AymmC, 0PrRYBCP1yGJ8xcBPmWhl5 | 17 | 27-147-574-9335 | 6819.74 | BUILDING | among the slyly regular theodolites kindle blithely courts. carefully even theodolites haggle slyly alon
9 | 10 | 10 | Customer#000000010 | 6LrEaV6KR6PLVcgl2ArL Q3rqzLzcT1 v2 | 5 | 15-741-346-9870 | 2753.54 | HOUSEHOLD | es regular deposits haggle. fur
(5 rows)
presto:tpch> SELECT sum(account_balance) FROM customer LIMIT 10;
_col0
------------
6681865.59
(1 row)
現在 customer 主題訊息中的所有欄位都可作為 Presto 資料表欄位使用。
步驟 7：使用即時資料¶
Presto 可以在 Kafka 中即時查詢送達的資料。為了模擬即時資料饋送，本教學將設定將即時推文饋送到 Kafka 中。
設定即時 Twitter 訊息¶
下載 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
在https://dev.twitter.com/ 建立開發人員帳戶，並設定存取和消費者權杖。
建立
twistr.properties檔案，並將存取和消費者金鑰與密碼放入其中
twistr.access-token-key=...
twistr.access-token-secret=...
twistr.consumer-key=...
twistr.consumer-secret=...
twistr.kafka.brokers=localhost:9092
在 Presto 上建立推文資料表¶
將推文資料表新增至 etc/catalog/kafka.properties 檔案
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,tweets
kafka.hide-internal-columns=false
為 Twitter 訊息新增主題定義檔案，如同 etc/kafka/tweets.json
{
"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"
}
]
}
}
由於此資料表沒有明確的結構描述名稱，因此會將其放入 default 結構描述中。
饋送即時資料¶
啟動 twistr 工具
$ java -Dness.config.location=file:$(pwd) -Dness.config=twistr -jar ./twistr
twistr 連線到 Twitter API，並將「範例推文」訊息饋送到名為 twitter_feed 的 Kafka 主題中。
現在針對即時資料執行查詢
$ ./presto-cli --catalog kafka --schema default
presto:default> SELECT count(*) FROM tweets;
_col0
-------
4467
(1 row)
presto:default> SELECT count(*) FROM tweets;
_col0
-------
4517
(1 row)
presto:default> SELECT count(*) FROM tweets;
_col0
-------
4572
(1 row)
presto: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
494227746214535169 | gu8tn | ja | 2014-07-29 14:07:31.000
494227746219126785 | pequitamedicen | es | 2014-07-29 14:07:31.000
494227746201931777 | josnyS | ht | 2014-07-29 14:07:31.000
494227746219110401 | Cafe510 | en | 2014-07-29 14:07:31.000
494227746210332673 | Da_JuanAnd_Only | en | 2014-07-29 14:07:31.000
494227746193956865 | Smile_Kidrauhl6 | pt | 2014-07-29 14:07:31.000
494227750426017793 | CashforeverCD | en | 2014-07-29 14:07:32.000
494227750396653569 | FilmArsivimiz | tr | 2014-07-29 14:07:32.000
494227750388256769 | jmolas | es | 2014-07-29 14:07:32.000
(10 rows)
現在，有一個可以透過 Presto 查詢的 Kafka 即時訊息。
後記：時間戳記¶
在上一步設定的推文（tweets）feed 中，每則推文的 created_at 屬性都包含一個 RFC 2822 格式的時間戳記。
presto: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
Tue Jul 29 21:07:32 +0000 2014
Tue Jul 29 21:07:33 +0000 2014
Tue Jul 29 21:07:34 +0000 2014
Tue Jul 29 21:07:35 +0000 2014
(5 rows)
推文表格的主題定義檔案包含使用 rfc2822 轉換器對應到時間戳記的設定。
...
{
"name": "created_at",
"mapping": "created_at",
"type": "TIMESTAMP",
"dataFormat": "rfc2822"
},
...
這使得原始資料可以對應到 Presto 的時間戳記欄位。
presto: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
2014-07-29 14:07:21.000 | Tue Jul 29 21:07:21 +0000 2014
2014-07-29 14:07:22.000 | Tue Jul 29 21:07:22 +0000 2014
2014-07-29 14:07:23.000 | Tue Jul 29 21:07:23 +0000 2014
2014-07-29 14:07:24.000 | Tue Jul 29 21:07:24 +0000 2014
(5 rows)
Kafka 連接器包含 ISO 8601、RFC 2822 文字格式以及使用自 epoch 以來的秒或毫秒數的數值時間戳記的轉換器。還有一個通用的、基於文字的格式化工具，它使用 Joda-Time 格式字串來解析文字欄位。