TP2 : Manipulation des données avec Apache Hive

Objectifs du TP : Au terme de ce TP, vous serez capable de :

• Manipuler des commandes HQL (Hive Query Language) qui vous permettront de créer, charger, modifier, supprimer les données
• Préparer les données avec Hive pour qu’elles soient exploitées par le traitement par lot (batch processing) avec Hadoop MapReduce

Hive est un système d’entrepôt des données (Datawarehouse) qui est utilisé pour le requêtage et l’analyse des larges datasets stockées dans HDFS. Il se base sur MapReduce pour effectuer ses traitements.

Dans ce TP, nous allons découvrir hive. Lancez hadoop et ses services si vous venez de vous connecter à votre machine:

start-all.sh

puis tapez:

hive

Vous aurez une ligne de commande qui commence par :

hive>

Si vous voulez sortir vous tapez :

quit;

Des erreurs rencontrées Pour accéder à Hive cli pour la toute première fois sur votre machine, vous devez lancer le service:

hive --service metastore &

Si on t’affiche l’erreur de Namenode is in Safe mode, lancez cette commande:

hdfs dfsadmin -safemode leave

Row format

Nous découvrons dans cette section, comment créer une table avec des champs délimités par ‘,’ (Remarque : dans l’affichage de la table, vous ne verrez pas les ‘,’, c’est au niveau du fichier physique que les données seront séparées par ‘,’).

Faites des captures écran pour garder la trace de votre travail pour la validation ! Si une commande de création de table ou de base ne marche pas car la table/base existe, appelez **DROP DATABASE ** ; ou **DROP TABLE ;**

Tapez :

CREATE TABLE jobs 
(id INT, title STRING, salary INT, posted TIMESTAMP) 
ROW FORMAT DELIMITED 
FIELDS TERMINATED BY ',';

Vous avez créé une table sous la base default. Vérifiez l’existence de cette table avec la commande:

SHOW TABLES;

Remplissez cette table avec:

INSERT INTO jobs 
VALUES (1,'Data Analyst',135000,'2016-12-21 15:52:03');

Remarquez le lancement d’un job MapReduce pour faire cette insertion! Pour une insertion, la latence est forte. Visualisez cette ligne bien enregistrée dans votre table en tapant:

SELECT * FROM JOBS;

Quittez hive cli avec quit ;

quit;

Le répertoire /user/hive/warehouse est le répertoire que hive utilise pour stocker ses bases de données et ses tables par défaut. Visualisez le contenu de ce répertoire:

hdfs dfs -ls  /user/hive/warehouse

Vous devez voir votre table jobs sous forme d’un répertoire. Affichez encore le contenu de ce répertoire:

hdfs dfs -ls  /user/hive/warehouse/jobs

Le fichier 000000_0 contient les lignes de votre table? Affichez le contenu du fichier à l’aide de la commande

hdfs dfs -cat /user/hive/warehouse/jobs/000000_0

Notez que les valeurs de chaîne et d’horodatage stockées dans ce fichier ne sont pas placées entre guillemets. Des guillemets ont été utilisés dans l’instruction INSERT pour encadrer la chaîne littérale et les valeurs d’horodatage, mais Hive ne stocke pas ces guillemets dans les fichiers de données de la table. Reconnectez-vous à hive cli avec la commande hive.

hive

Exécutez maintenant :

CREATE TABLE jobs_tsv 
(id INT, title STRING, salary INT, posted TIMESTAMP) 
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

Insérez une ligne dans cette table:

INSERT INTO jobs_tsv 
VALUES (1,'Data Analyst',135000,'2016-12-21 15:52:03');

Quittez hive :

quit;

Examinez le contenu du fichier /user/hive/warehouse/jobs_tsv pour voir la différence entre les deux délimiteurs ‘,’ et ‘\t’ en visualisant le contenu du fichier sous le répertoire jobs_tsv.

hdfs dfs -cat /user/hive/warehouse/jobs_tsv/000000_0

Dans Hive, les délimiteurs , (virgule) et \t (tabulation) sont utilisés pour séparer les colonnes lors de l’importation ou de l’exportation de données dans des fichiers. On préfère l’usage du délimiteur tabulation quand les données elles-mêmes contiennent des virgule car la tabulation est moins susceptible d’y apparaître. Si vous ne précisez pas le délimiteur, hive éventuellement considérera que vos données sont séparées par des espaces ou des tabulations.

Table externe

Nous avons vu que hive stocke les tables dans un répertoire par défaut. Nous allons, voir dans cette manipulation, que hive peut stocker des tables en dehors de ce répertoire, mais dans ce cas, il les considère comme des tables externes. La spécificité de celles-ci est qu’elles ne seront pas supprimées du disque en lançant la commande DROP TABLE, mais seulement elle seront inacessibles depuis hive. Ces tables sont particulièrement utiles lorsqu’on souhaite conserver les données sur le disque pour d’autres usages ou lorsqu’elles sont partagées avec d’autres systèmes.

Créez une table EXTERNE avec la commande suivante :

CREATE EXTERNAL TABLE default.investors 
(name STRING, amount INT, share DECIMAL(4,3)) 
ROW FORMAT DELIMITED 
FIELDS TERMINATED BY ',';

Tapez:

INSERT INTO default.investors (name, amount, share)
VALUES ('Karim Beguir', 10000, 0.325);

Vérifiez qu’elle ne figure plus parmi les autres tables :

show tables;

Quittez hive en tapant :

quit;

Visualisz le contenu du répertoire de hive et vous verrez que la table existe quand même.

hdfs dfs -ls /user/hive/warehouse

Effacez la table:

DROP TABLE default.investors;

Re-chargez la table avec:

CREATE EXTERNAL TABLE default.investors(name STRING, amount INT, share DECIMAL(4,3));

Observez le résultat:

SELECT * FROM investors ;

Stored as

Créez une nouvelle table avec :

CREATE TABLE jobs_txt (id INT, title STRING, salary INT, posted TIMESTAMP) 
STORED AS TEXTFILE;

Insérez une ligne dans la table :

INSERT INTO jobs_txt VALUES (1,'Data Analyst',135000,'2016-12-21 15:52:03');

Examinez le fichier sous le répertoire /user/hive/warehouse/jobs_txt . Remarquez que vous pouvez clairement voir le contenu du fichier et ses valeurs dedans.

Maintenant, on va créer un autre format, le format « parquet » :

CREATE TABLE jobs_parquet (id INT, title STRING, salary INT, posted TIMESTAMP) 
STORED AS PARQUET ;

Insérez une ligne dans la table :

INSERT INTO jobs_parquet VALUES (1,'Data Analyst',135000,'2016-12-21 15:52:03');

Examinez le fichier sous le répertoire /user/hive/warehouse/jobs_parquet. Remarquez que vous ne voyez pas un contenu cohérent mais plutôt il y a du code ASCII dedans !

Location

Créez la table :

CREATE TABLE jobs_training 
(id INT, title STRING, salary INT, posted TIMESTAMP) 
LOCATION '/user/training/jobs_training/';

Vérifiez la création du répertoire «/user/training/jobs_training » et que le dossier est vide jusqu’à présent.

INSERT INTO jobs_training 
VALUES (1,'Data Analyst',135000,'2016-12-21 15:52:03');

Regardez une autre fois dans le répertoire de la table et vérifiez si vous avez bien ajouté cette ligne.

If not exits and LIKE

Nous pouvons créer une table ayant le même schéma qu’une table existante. Tapez:

CREATE TABLE new_investors LIKE investors;

Si vous tentez d’exécuter cette même commande une autre fois, vous obtiendrez une erreur! Pour éviter que celà puisse apparaître, vous utilisez IF NOT EXISTS:

CREATE TABLE new_investors LIKE investors;

Tapez :

INSERT INTO new_investors SELECT * from investors;

et observez le contenu de la table new_investors:

SELECT * FROM new_investors;

Table description

Tapez :

DESCRIBE jobs_txt ;
DESCRIBE FORMATTED jobs_txt ;

Si vous obtenez la valeur MANAGED_TABLE, c’est parce que la table n’a pas été créée en tant que EXTERNAL.

Table modification

Beaucoup de modifications sont possibles avec ALTER. Modifions une table interne par une table externe :

ALTER TABLE investors SET TBLPROPERTIES('EXTERNAL'='TRUE');

On peut aussi changer son nom:

ALTER TABLE investors RENAME TO companies ;

Également, on peut changer la base de données hôte :

CREATE DATABASE IF NOT EXISTS section;
ALTER TABLE default.companies RENAME TO section.companies;

Vous pouvez également changer les types des champs avec ALTER.

Partitioning

Dans hive, nous pouvons effectuer deux types de partitionnement des données : statique et dynamique. Le partitionnement statique dans Hive signifie que les partitions sont explicitement spécifiées par l’utilisateur lors de l’insertion des données dans une table partitionnée. L’utilisateur doit définir les valeurs de partition pour chaque partition dans la requête INSERT. Les partitions sont donc créées au moment de l’insertion. Par exemple:

CREATE TABLE sales (
    product STRING,
    amount INT
)
PARTITIONED BY (year INT, month INT);

-- Insertion de données dans une partition spécifique (année 2023, mois 1)
INSERT INTO sales PARTITION (year=2023, month=1)
VALUES ('product1', 100);

Quant au partionnement dynamique, il est recommandé pour traiter une grande quantité de données à insérer et où les partitions peuvent être différentes pour chaque lot de données. Ainsi, les partitions seront créées d’une manière automatique. Prenons un exemple:

Quittez hive

quit;

Téléchargez customers.csv à partir du lien:

https://drive.google.com/drive/folders/1yHi5gPFu_4999GhdemwRD8ux2mllufU8?usp=drive_link

Tapez:

mv Downloads/customers.csv customers.csv

Chargerons le fichier customers.csv sur HDFS avec la commande :

hdfs dfs -put customers.csv /user/hive/warehouse/customers.csv

Accédez maintenant à hive CLI:

hive

Créez la table customers où on chargera le fichier customers.csv. Remarquez le champ TBLPROPERTIES (“skip.header.line.count”=”1”); indique que nous ne voulons par charger le « header » du fichier source :

CREATE TABLE IF NOT EXISTS customers (
    id_cust INT,
    name STRING,
    country STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
TBLPROPERTIES ("skip.header.line.count"="1");

Maintenant, nous chargerons le fichier dans la table :

LOAD DATA INPATH '/user/hive/warehouse/customers.csv' INTO TABLE customers;

Vérifiez le contenu de la table :

SELECT * FROM customers ;

Nous allons maintenant créer des partitions de cette table de façon à avoir des fragments de la base suivant un certain critère sous le dossier /user/hive/warehouse comme indiqué dans la figure :

Pour ce faire, préparez cette opération de partition, en créant tout d’abord la table customers_by_country :

CREATE TABLE customers_by_country
(id_cust INT, name STRING)
PARTITIONED BY (country string)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ',';

Tapez ensuite pour demander à hive d’opter pour la partition dynamique :

set hive.exec.dynamic.partition.mode=nonstrict;

Puis:

INSERT OVERWRITE TABLE customers_by_country
PARTITION(country)
SELECT id_cust ,name , country FROM customers;

Sortez de hive cli avec quit; et puis lancez la commande :

hdfs dfs -ls /user/hive/warehouse/customers_by_country

Manipulation : Pipeline Batch Processing Data

Le but de cette manipulation est la simulation d’extraction des données d’une base de données relationnelles vers hive afin de le traiter ensuite à l’aide de hadoop MapReduce comme est indiqué dans la figure suivante:

1. Pour ce faire,1. Connectez-vous au conteneur hive-server:

   docker exec -it hive-server bash
   

2. Une fois connecté, nous allons lancer un script pour configurer sqoop.

   chmod 7 start-hive.sh
   

   Puis lancez : (si vous le lancez pour la première fois, vous aurez un message : rm: `/user/sqoop’: No such file or directory)

   ./start-hive.sh
   

   La table à créer s’appelle order_detail

3. Connectz-vous à postgres et visualisez les bases de données qui sont dedans: cmd sudo -i -u postgres psql `
4. Créer avec hive cli la base de données lab2_big_data. ``cmd hive

   Puis:
   ``cmd
   CREATE DATABASE lab2_big_data;
   

5. Accédez au container de postgres avec la commande :

   sudo docker exec -it postgres-db psql -U postgres
   

6. Afficher toutes les bases de données que vous avez à bord avec la commande :

   \l
   

7. Vous allez accéder à la base « temp_db » avec la commande :

   \c temp_db
   

8. Vous pouvez vérifier s’il y a des tables dans la base :

   \dt
   

9. Nous allons créer une table qui s’appelle order_detail que nous chargerons d’un fichier csv :

   DROP TABLE IF EXISTS order_detail;
   

   Ensuite:

   CREATE TABLE IF NOT EXISTS order_detail (
   order_created_timestamp TIMESTAMP,
   status VARCHAR,
   price INT,
   discount FLOAT,
   id VARCHAR PRIMARY KEY NOT NULL,
   driver_id VARCHAR,
   user_id VARCHAR,
   restaurant_id VARCHAR
   );
   

   Puis :

   COPY order_detail FROM '/var/lib/postgresql/data/lmwn/order_detail.csv' DELIMITER ',' CSV HEADER;
   

   Vous venez avec cette commande de charger les données dans le fichier .csv dans la table de base de données sous postgres container. Quittez votre container en tapant:

   \q
   

10. Nous allons maintenant importer les données de la table que nous venons de créer sur notre RDBMS vers Hive à l’aide de Sqoop. Tapez la commande suivante :

    docker exec -it hive-server bash
    

    puis à l’intérieur du container, vous lancez:

    /usr/lib/sqoop/bin/sqoop import --connect jdbc:postgresql://database:5432/temp_db --table order_detail --username postgres --password passw0rd --hive-import --hive-table lab2_big_data.order_detail_hive --num-mappers 1 --map-column-hive order_created_timestamp=STRING,status=STRING,price=INT,discount=FLOAT,id=STRING,driver_id=STRING,user_id=STRING,restaurant_id=STRING
    

    Accédez à « localhost :8088 » pour voir le succès de votre transfert de données avec sqoop. La raison pour laquelle vous trouvez ce transfert sur le history server de hadoop est le fait que Hive a déclenché un job MapReduce pour faire cette opération !

Vérifier le succès du transfert en revenant vers hive:

SELECT * FROM lab2_big_data.order_detail_hive LIMIT 5;

A vous de jouer!

1. Partitionnez vos données par id_restaurant.
2. Ecrivez un programme de MapReduce qui lit de /user/hive/warehouse/ qui somme le nombre de clients servis (user_id) pour un certain restaurant.