Apache Presto - Kafka 聯結器

Presto 的 Kafka 聯結器允許使用 Presto 訪問 Apache Kafka 中的資料。

前提條件

下載並安裝以下 Apache 專案的最新版本。

Apache ZooKeeper
Apache Kafka

啟動 ZooKeeper

使用以下命令啟動 ZooKeeper 伺服器。

$ bin/zookeeper-server-start.sh config/zookeeper.properties

現在，ZooKeeper 在 2181 埠啟動。

啟動 Kafka

在另一個終端中使用以下命令啟動 Kafka。

$ bin/kafka-server-start.sh config/server.properties

Kafka 啟動後，它使用埠號 9092。

TPCH 資料

下載 tpch-kafka

$  curl -o kafka-tpch 
https://repo1.maven.org/maven2/de/softwareforge/kafka_tpch_0811/1.0/kafka_tpch_ 
0811-1.0.sh

現在，您已使用上述命令從 Maven 中央倉庫下載了載入程式。您將收到類似以下的響應。

% Total    % Received % Xferd  Average Speed   Time    Time     Time  Current 
                                 Dload  Upload   Total   Spent    Left  Speed 
  0     0    0     0    0     0      0      0 --:--:--  0:00:01 --:--:--     0  
  5 21.6M    5 1279k    0     0  83898      0  0:04:30  0:00:15  0:04:15  129k
  6 21.6M    6 1407k    0     0  86656      0  0:04:21  0:00:16  0:04:05  131k  
 24 21.6M   24 5439k    0     0   124k      0  0:02:57  0:00:43  0:02:14  175k 
 24 21.6M   24 5439k    0     0   124k      0  0:02:58  0:00:43  0:02:15  160k 
 25 21.6M   25 5736k    0     0   128k      0  0:02:52  0:00:44  0:02:08  181k 
 ………………………..

然後，使用以下命令使其可執行，

$ chmod 755 kafka-tpch

執行 tpch-kafka

使用以下命令執行 kafka-tpch 程式，以使用 tpch 資料預載入多個主題。

查詢

$ ./kafka-tpch load --brokers localhost:9092 --prefix tpch. --tpch-type tiny

結果

2016-07-13T16:15:52.083+0530 INFO main io.airlift.log.Logging Logging 
to stderr
2016-07-13T16:15:52.124+0530 INFO main de.softwareforge.kafka.LoadCommand
Processing tables: [customer, orders, lineitem, part, partsupp, supplier,
nation, region]
2016-07-13T16:15:52.834+0530 INFO pool-1-thread-1
de.softwareforge.kafka.LoadCommand Loading table 'customer' into topic 'tpch.customer'...
2016-07-13T16:15:52.834+0530 INFO pool-1-thread-2
de.softwareforge.kafka.LoadCommand Loading table 'orders' into topic 'tpch.orders'...
2016-07-13T16:15:52.834+0530 INFO pool-1-thread-3
de.softwareforge.kafka.LoadCommand Loading table 'lineitem' into topic 'tpch.lineitem'...
2016-07-13T16:15:52.834+0530 INFO pool-1-thread-4
de.softwareforge.kafka.LoadCommand Loading table 'part' into topic 'tpch.part'...
………………………
……………………….

現在，使用 tpch 載入了 Kafka 表 customers、orders、supplier 等。

新增配置設定

讓我們在 Presto 伺服器上新增以下 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

在上述配置中，Kafka 表使用 Kafka-tpch 程式載入。

啟動 Presto CLI

使用以下命令啟動 Presto CLI：

$ ./presto --server localhost:8080 --catalog kafka —schema tpch;

這裡“tpch”是 Kafka 聯結器的模式，您將收到以下響應。

presto:tpch>

列出表

以下查詢列出“tpch”模式中的所有表。

查詢

presto:tpch> show tables;

結果

  Table 
---------- 
 customer 
 lineitem 
 nation 
 orders
 part 
 partsupp 
 region 
 supplier

描述 Customer 表

以下查詢描述“customer”表。

查詢

presto:tpch> describe customer;

結果

  Column           |  Type   |                   Comment 
-------------------+---------+--------------------------------------------- 
 _partition_id     | bigint  | Partition Id 
 _partition_offset | bigint  | Offset for the message within the partition 
 _segment_start    | bigint  | Segment start offset 
 _segment_end      | bigint  | Segment end offset 
 _segment_count    | bigint  | Running message count per segment 
 _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

列印頁面