Apache Presto - Connecteur KAFKA

Le connecteur Kafka pour Presto permet d'accéder aux données d'Apache Kafka en utilisant Presto.

Conditions préalables

Téléchargez et installez la dernière version des projets Apache suivants.

Apache ZooKeeper
Apache Kafka

Démarrez ZooKeeper

Démarrez le serveur ZooKeeper à l'aide de la commande suivante.

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

Maintenant, ZooKeeper démarre le port sur 2181.

Démarrez Kafka

Démarrez Kafka dans un autre terminal à l'aide de la commande suivante.

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

Après le démarrage de kafka, il utilise le numéro de port 9092.

Données TPCH

Télécharger tpch-kafka

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

Vous avez maintenant téléchargé le chargeur depuis Maven central en utilisant la commande ci-dessus. Vous obtiendrez une réponse similaire à la suivante.

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

Ensuite, rendez-le exécutable à l'aide de la commande suivante,

$ chmod 755 kafka-tpch

Exécutez tpch-kafka

Exécutez le programme kafka-tpch pour précharger un certain nombre de rubriques avec des données tpch à l'aide de la commande suivante.

Requete

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

Résultat

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

Désormais, les clients, commandes, fournisseurs, etc. des tables Kafka sont chargés à l'aide de tpch.

Ajouter des paramètres de configuration

Ajoutons les paramètres de configuration du connecteur Kafka suivants sur le serveur Presto.

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

Dans la configuration ci-dessus, les tables Kafka sont chargées à l'aide du programme Kafka-tpch.

Démarrez Presto CLI

Démarrez Presto CLI à l'aide de la commande suivante,

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

Ici “tpch" est un schéma pour le connecteur Kafka et vous recevrez une réponse comme suit.

presto:tpch>

Tables de liste

La requête suivante répertorie toutes les tables de “tpch” schéma.

Requete

presto:tpch> show tables;

Résultat

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

Décrire la table des clients

La requête suivante décrit “customer” table.

Requete

presto:tpch> describe customer;

Résultat

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