Téléchargement de jeux de données dans HDFS et Hive

Introduction

De nos jours, l’analyse de larges quantités de données devient de plus en plus possible grâce aux technologies du Big data (Hadoop, Spark,…). Cela explique l’explosion du volume de données et la présence de nombreuses sources de données disponibles telles que celles répertoriées dans le site Github Awesome Public Datasets. Cette disponibilité me permit la réalisation de plusieurs tests sur différents types de données pertinentes. C’est le cas d’un test de comparasion de différents formats de fichiers (CSV, JSON,…) où nous avons isolé plusieurs types de données respectivement dans Hadoop HDFS et Hive.

À travers cet article, nous avons trouvé intéressant de partager notre script pour charger les jeux de données. Nous décrirons comment les télécharger efficacement dans HDFS et le prétraitement requis pour les exposer dans Hive. Par souci de clarté, l’accent est mis sur un exemple de jeu de données, celui des Taxis de New York. N’hésitez pas à ajuster les paramètres en fonction du jeu de données que vous ciblez.

Chargement dans HDFS

Nous avons créé un script bash pour charger les données dans HDFS. Le jeu de données des Taxis de la ville de New York est composé de plusieurs fichiers organisés par dates. Voici un exemple d’URL : https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2010-01.csv.

Pour chaque URL présente dans le fichier datasets_to_download, le script télécharge le contenu avec curl et le redirige vers la commande HDFS put. De cette façon, aucun enregistrement n’est stocké sur la machine hôte exécutant la commande, l’empêchant de saturer. Ceci est important car cette machine hôte n’est souvent pas dimensionnée pour stocker une grande quantité de données. Par exemple, lors de nos enseignements, nous demandons souvent à nos étudiants de se connecter à un edge node, qui est un conteneur Linux, et de leur demander de télécharger un jeu de données, déclenchant plus de 40 téléchargements en parallèle.

#!/bin/bash
username=`whoami`
hostname="cluster_hostname_or_ip"
for line in $(cat ./datasets_to_download)
do
  # The filename variable retrieves the name of the data that is usually present at the end of the download link.
  filename =`awk -F"/" '{print $NF}' <<< "${line}"`
  dir = "/user/${username}/taxi_trip/staging"
  sshpass -f <(printf '%s\n' your_password) ssh $user@$hostname \
  "hdfs dfs -put <(curl -sS $line) $dir/$filename"
done

REMARQUE : le script shell doit être exécutable par l’utilisateur, par exemple avec chmod u + x scripts.sh.

Parfois, vous pouvez traiter un lien qui n’expose pas le nom des données dans son nom. Par exemple, l’Université de l’Illinois fournit une version légèrement améliorée des données des taxis de New York. Les URL de téléchargement ressemblent à :

https://public.boxcloud.com/d/1/b1!wgmpX-s_1eUYtZ1bbQNuWD-BOpAxq8NzBk0Vg-qGOdsU9GyqlKu_iLszh_UaZey6yA8bjcz7sYVjNH3JCrii-lb6VPcotyAnlsTZJHzLPl7zvvon6W2Hh-PolWX0Iz-ZCT4LZGvhpi0p-F5_cIbuyQ3ZjW7jtVR11DMrqs1QESdUKjeoaCyCFnkVv9QkASQfF0zQKhYCdhPX8_5a7-_plD3NnxQW4WNwqXJQv4OK-J-oFfkAjcRBNpxu3-zLtl4v9QafdaSEICun4K3FIlOovwmtfhRTlzN9mywnf7e46qNbqPO8Zjbrt3GgPxp6L5ZviSec6RcgaRgu2O940ZvT_7i4u95s3XOlf3sIib8Gw8Xw8cr0juDpYM228BCSZtdBqL1pO_CFjNeVz7Fwj1R0vKCuYUEf7pM9igDUAJNClysA-RaMdBqFKphphd0m_dM8Vm2g8PhoLCEgBX9l04Qzdgwyk1ckebpE2Nb64hqHYHuO0VLiRUZJEw6dCpufKgvv5iHf6YNTuFXNuNlJKdU2cQKMIamG94xW510FYWTari-EqfbKVapYF9hKOOldE0ex4tOnVhAPPhpzRKk-g1UfVCsOqzhV38__U3OuzSNJSdq9oq1_PMw4ZMle6AlBaGfyXfz3cBwgQLutW_dXIQaibouAdyJ7n-2sXhnDyd8UeEORwkYehv64-rMmj6mZ4FreJl_YZQ_bIc54aYkYQz2A3-VzAdg66KXD6sEIr2pMsrK2l3cT9MeY2DshgVLl0_zIoZlL2GT2grLna3VFT4iw67kIFQmviYa5Uxm6lzoFsdaVocO9zOUyMhR6k5E4hWkHpFflwT9XSBY6s_Es6YaOFuCJLODIs0hKUm9bGa2CvvexqNnUr_RAZVqkwL25z7BClrq-nPcscUAYIYbE9E1g3Qdw8g1pksh4UQ_-YAuIKo3xAIqzXt0cV6ao5l63ldksBxyCH9v4wtp4BZvRm_cnaT29s7ppDEXyArxOcPW5CbpVfz8L-8H5sXzXmuDPbRASChlX4QtHTX83BpGN1p-fcj_52Ob4UGPXzxdA6KR5CZ9uZFK4GACEKiZxxbQNppcL-l3zd_iaY7w45seYmCvx6SA44xOEDDzKNVnkA4g3LP47dR4Msbi-NMJHJFXeI-1gywM1krey_fKCmjKe_dKP6j2n7WeyYlyS1kBqpQYSRC8chUcgGKCz_Guc4YaZ7n79NZP6zUOIMbWQOGwxpvVeCduKim7f0VJlNV3kgi71xwQ./download

Dans ce cas de figure, le nom des données de chaque lien de teléchargement pourrait être défini comme suit, avec un index suffixé au nom de fichier :

#!/bin/bash
username=`whoami`
hostname="cluster_hostname_or_ip"
for line in $(cat ./datasets_to_download)
do
  let i=i+1
  filename="trip_data_${i}.zip"
  dir= "/user/${username}/taxi_trip/staging"
  sshpass -f <(printf '%s\n' your_password) ssh $user@$hostname \
  "hdfs dfs -put <(curl -sS $line) $dir/$filename"
done

Dans le cas de données compressées au format gzip, les fichiers peuvent être décompressés avant leur dépôt dans HDFS avec la commande gzip -d :

unzip_filename=$"${name%.*}"_"csv"
hdfs dfs -put <(curl -sS $line -L | gzip -d) $dir/$unzip_filename

Déclaration dans Hive

La déclaration de données dans Hive implique la création de leur schéma. Dans notre cas, nous avons téléchargé des fichiers CSV où les enregistrements sont séparés par \n (séparateur de ligne Unix) et où les champs sont séparés par , (virgule). Chaque fichier correspond à une année de données et ils se trouvent tous dans le même dossier HDFS. Nous pouvons tirer parti du fonctionnement internes de Hive qui associent une table à un répertoire HDFS, non à un fichier HDFS, et considérer tous les fichiers de ce répertoire comme le jeu de données complet.

La requête suivante crée une table Hive qui combine les données de trajets en taxi de chaque année. Remplacez la valeur LOCATION par le chemin HDFS stockant vos fichiers téléchargés. Le tableau est temporaire, nous l’utiliserons pour créer un tableau plus optimisé plus tard et l’effacerons une fois terminé.

CREATE EXTERNAL TABLE taxi_trip_staging
(
  medallion Numeric, hack_license Numeric,
  vendor_id String,  rate_code Numeric,
  store_and_fwd_flag Numeric, pickup_datetime string, 
  dropoff_datetime string, passenger_count Numeric,
  trip_time_in_secs Numeric, trip_distance Numeric,
  pickup_longitude Numeric, pickup_latitude Numeric, 
  dropoff_longitude Numeric, dropoff_latitude Numeric
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
STORED AS textfile
LOCATION '/user/${env:USER}/taxi_trip/staging'
TBLPROPERTIES("skip.header.line.count"="1");

Le jeu de données est désormais disponible depuis Hive. Nous allons créer une deuxième table optimisée afin de profiter du partitionnement. Les données seront partitionnées par année et par mois. La première partie de la requête crée une nouvelle table partitionnée et la seconde partie charge cette table à partir des données de la table non partitionnée taxi_trip_staging crée précédemment.

CREATE EXTERNAL TABLE taxi_trip
(
 medallion Numeric, hack_license Numeric,
 vendor_id String,  rate_code Numeric,
 store_and_fwd_flag Numeric,pickup_datetime string,
 dropoff_datetime string,  passenger_count Numeric,
 trip_time_in_secs Numeric, trip_distance Numeric,
 pickup_longitude Numeric, pickup_latitude Numeric ,
 dropoff_longitude Numeric, dropoff_latitude Numeric
)
PARTITIONED BY (year string, month string)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
STORED AS textfile
LOCATION '/user/${env:USER}/taxi_trip/gold';
-- Load the data
INSERT OVERWRITE TABLE taxi_trip PARTITION(year, month)
SELECT
 medallion,hack_license, vendor_id, rate_code, store_and_fwd_flag,
 pickup_datetime, dropoff_datetime, passenger_count, trip_time_in_secs,
 trip_distance, pickup_longitude, pickup_latitude, dropoff_longitude,
 dropoff_latitude, year(pickup_datetime), month(pickup_datetime)
FROM taxi_trip_staging;
-- Remove the temporary table (not its data)
DROP TABLE taxi_trip_staging;

Les données sont désormais accessibles via Hive et vous pouvez tirer profit des avantages qu’elle offre.

Conclusion

J’espère que le script que nous avons partagé et ses recommandations vous aideront à charger efficacement les données dans HDFS et Hive. L’utilisation de la mémoire et du disque restera faible sur l’hôte exécutant la commande. Une fois les données chargées dans Hive, vous pouvez bénéficier de sa rapidité et de sa facilité pour vos besoins de traitement de données (synthèse de données, requête ad-hoc, …).