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 Central에서 로더를 다운로드했습니다. 다음과 유사한 응답을 받게됩니다.

% 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'...
………………………
……………………….

이제 Kafka 테이블 고객, 주문, 공급 업체 등은 tpch를 사용하여로드됩니다.

구성 설정 추가

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” 표.

질문

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