Apache Hive 3, nouvelles fonctionnalités et conseils et astuces

Disponible depuis juillet 2018 avec HDP3 (Hortonworks Data Platform 3), Apache Hive 3 apporte de nombreuses fonctionnalités intéressantes à l’entrepôt de données. Malheureusement, comme beaucoup de versions majeures de logiciels “FOSS”, elle contient quelques écueils et peu de documentation.

Je vais d’abord passer en revue les nouvelles fonctionnalités disponibles avec Hive 3, puis donner quelques conseils et astuces appris avec son utilisation en production pendant plusieurs mois. La première partie de l’article est basée sur la présentation What’s new in Apache Hive donnée par Jason Dere au DataWorks Summit 2019 Barcelona.

Les nouvelles fonctionnalités de Hive 3

De nombreuses fonctionnalités et améliorations ont été apportées à Hive 3, élargissant considérablement les cas d’utilisation pouvant être couverts. Passons en revue celles qui sont disponibles dans HDP3, c’est le cas de la plupart d’entre elles.

Les vues matérialisées (MV)

Les vues matérialisées sont enfin disponibles dans HDP. Meilleures que des vues traditionnelles, elle apportent :

• Un stockage pour les résultats intermédiaires : comme leur nom l’indique, les résultats des vues matérialisées sont stockés, ce qui permet une mutualisation des coûts de calcul ;
• La reconstruction incrémentale : la mise à jour d’une vue matérialisée se fait uniquement à partir des nouvelles données insérées dans les tables sources après la dernière mise à jour ;
• La réécriture de requêtes : quand c’est possible, l’optimiseur de coûts utilise les vues matérialisées existantes au lieu des tables sources pour planifier et donc optimiser les requêtes, de manière. transparente, sans intervention de l’utilisateur !

Pour plus d’informations sur les vues matérialisées, consultez l’article de mon collègue Paul-Adrien Cordonnier Accélérer vos requêtes avec les vues matérialisées dans Apache Hive.

Fonctionnalités SQL

Contraintes and valeurs par défaut

Il est maintenant possible d’utiliser les contraintes SQL UNIQUE et NOT NULL, en plus de PRIMARY|FOREIGN KEY (ajoutées dans Hive 2.1). De plus, vous pouvez spécifier une valeur DEFAULT pour chaque colonne et utiliser cette valeur par défaut dans les instructions INSERT et UPDATE.

CREATE TABLE users (
  id INT NOT NULL,
  firstname STRING,
  lastname STRING,
  join_date DATE DEFAULT CURRENT_DATE(),
  PRIMARY KEY (id) DISABLE NOVALIDATE
);
INSERT INTO TABLE users VALUES (1, 'John', 'Doe', DEFAULT);

Informations sur les tables

Hive 3 permet d’explorer facilement tout l’entrepôt avec les bases information_schema et sys :

SELECT table_schema, table_name
FROM information_schema.columns
WHERE column_name LIKE '%ssn%';

Connecteurs JDBC, Druid et Kafka

Deux nouveaux connecteurs sont disponibles pour les tables externes :

• Le connecteur JDBC (JdbcStorageHandler), qui est en lecture seule pour le moment. Il vous permet de lire/importer facilement des données depuis l’une des bases de données prises en charge : MySQL, PostgreSQL, Oracle, MSSQL et Apache Derby.
• Le connecteur Kafka, qui peut être utilisé pour requêter des données en temps réel à partir de Apache Kafka, et qui permet également l’insertion de données en temps réel vers Hive avec une garantie “exactly-once” et la transformation de données en temps réel ! Voir Integrating Hive and Kafka pour plus de détails.

De plus, l’ingestion “exactly-once” de Kafka vers Druid peut être gérée via Hive.

ACID v2

Dans les versions précédentes, Hive ACID (Atomicité, Cohérence, Isolation et Durabilité) a beaucoup apportées aux tables dites managed stockées dans le format Apache ORC :

• Ingestion de données en streaming ;
• Opérations INSERT/UPDATE/DELETE sur des table existantes sans impacter les opérations en cours.

La deuxième version de Hive ACID comporte plusieurs améliorations :

• Des performances aussi bonnes que les tables non-ACID ;
• Les tables transactionnelles (tables supportant les opérations ACID) ne doivent plus être obligatoirement bucketées ;
• La prise en charge des formats non-ORC pour les opérations INSERT/SELECT ;
• La compatibilité avec les fournisseurs de stockage Cloud, par exemple Amazon S3 ;
• La nouvelle API HiveStreaming (v2).

Hortonworks a décidé d’activer l’ACID par défaut dans HDP3. Par exemple, les tables ORC internes doivent être transactionnelles.

Malheureusement, la plupart des problèmes que j’ai rencontrés avec Hive 3 proviennent de Hive ACID et de l’API HiveStreaming. Ces problèmes et leur solutions sont couvert plus bas dans la deuxième partie de l’article.

Le Hive Warehouse Connector pour Apache Spark

À partir de HDP 3.0, toutes les interactions entre Hive et Apache Spark doivent passer par le connecteur Hive Warehouse Connector. Ce connecteur tire parti de Hive LLAP pour permettre une interaction streaming/ACID entre Hive et Spark. L’application Spark devra accéder à un Hive Server Interactive (avec LLAP activé) pour lire les tables managed de Hive, mais elle n’en aura pas besoin pour écrire dans des tables managed ou lire/écrire des tables externes. Ne soyez pas surpris si la manière traditionnelle d’accéder aux tables Hive depuis Spark ne fonctionne plus !

Gestion des workloads LLAP

Cette excellente nouvelle fonctionnalité améliore votre contrôle sur Hive LLAP (Live Long and Process) dans les environnements multi-tenant avec les ressource plans :

• Répartissez vos ressources LLAP en pools, par exemple bi (pour Business Analytics) et etl (pour Extract Transform and Load) ;
• Mapper automatiquement les applications aux pools, par exemple Tableau vers le pool bi ;
• Créez des triggers pour déplacer dynamiquement des applications d’un pool à un autre, par exemple déplacer les requêtes longues vers le pool etl ;
• Activer/désactiver vos ressource plans en fonction des besoins de vos utilisateurs (un actif à la fois).

CREATE RESOURCE PLAN my_plan;
CREATE POOL my_plan.bi
WITH ALLOC_FRACTION=70,QUERY_PARALLELISM=4;
CREATE POOL my_plan.etl
WITH ALLOC_FRACTION=30,QUERY_PARALLELISM=10;
CREATE TRIGGER my_plan.slow_query
WHEN execution_time_ms > 60000
DO MOVE TO etl;
ALTER PLAN my_plan SET DEFAULT POOL=bi;
ALTER PLAN my_plan ENABLE ACTIVATE;

Amélioration des performances

D’après un test de performance TPC-DS récent réalisé par l’équipe MR3, Hive LLAP 3.1.0 est le système SQL-sur-Hadoop le plus rapide disponible dans HDP 3.0.1. Le benchmark compare tous les systèmes SQL intégrés à HDP3 ainsi que Hive on MR3 (un nouveau moteur d’exécution pour Hadoop et Kubernetes) en exécutant un ensemble de 99 requêtes SQL.

Je vous invite à lire l’article très complet Performance Evaluation of SQL-on-Hadoop Systems using the TPC-DS Benchmark (edit : url indisponible) et les slides sur les performances de Jason Dere pour plus de détails sur ce qui rend Hive 3 plus rapide.

Hive 3, trucs et astuces

Je maintiens une instance de Hive 3 depuis plus de 5 mois maintenant. Au cours de cette période, j’ai rencontré des problèmes de stabilité et mis en œuvre des solutions parfois complexes. Plus de détails sur le setup Hive utilisé sont disponibles à la fin de l’article.

Si vous prévoyez d’utiliser Hive 3, cela vous aidera probablement à identifier vos problèmes et à les résoudre, au moins avant qu’ils ne soient résolus dans une prochaine version officielle.

Fuites de mémoire dans l’API HiveStreaming

La nouvelle API HiveStreaming mentionnée précédemment comporte deux fuites de mémoire connues (HIVE-20979). Cela impacte toutes les applications clientes utilisant l’API, qui risquent de planter à cause d’un problème de heap de la Java Virtual Machine.

Dans mon cas, l’application cliente était Apache NiFi et son processeur PutHive3Streaming. C’est le processeur officiel fourni avec NiFi 1.7.0 pour écrire vers Hive 3.0+. Comme il exploite l’API HiveStreaming (v2), mon cluster NiFi tombait toujours en panne après quelques heures de fonctionnement.

J’ai passé du temps à analyser NiFi et j’ai fini par réécrire le processeur (NiFi Hive3Streaming Fixed). Cette réimplémentation intègre les correctifs fournis par la communauté Hive (HIVE-20979). Si vous souhaitez en savoir plus sur ce problème particulier lié à NiFi, consultez l’article consacré Dépannage d’un pipeline NiFi-HiveStreaming (à venir).

Si vous regardez de près le ticket JIRA HIVE-20979 , vous verrez que le correctif est censé être dans la version 3.1.1 de Hive. Néanmoins, ce n’est pas le cas… Mais ce n’est pas si grave, car même si cela était corrigé, vous devriez attendre qu’Hortonworks migre vers Hive 3.1.1 et mettre à jour l’ensemble de votre cluster (si vous utilisez HDP).

Finalement, la bonne chose à propos de ce problème HiveStreaming est que vous pouvez y remédier. Comme je l’ai fait pour réécrire le processeur, étudiez les correctifs fournis par la communauté Hive (HIVE-20979) et intégrez les classes Java concernées dans votre application. N’hésitez pas à réutiliser des parties de mon code publié sur GitHub si vous en avez besoin !

Problèmes avec les compactions Hive ACID

Comme indiqué précédemment, les tables Hive sont transactionnelles par défaut dans HDP 3 et les tables managées en ORC ne peuvent pas être non transactionnelles. Cela signifie qu’elles supportent les opérations ACID :

• Atomicité : une opération réussit ou échoue ;
• Cohérence : une fois qu’une opération a été effectuée, chaque opération suivante verra ses résultats ;
• Isolation : une opération incomplète n’affecte pas les autres opérations ;
• Durabilité : une fois l’opération terminée, elle est préservée même en cas de panne de la machine ou du système.

Bien que cela donne plus de fonctionnalités aux tables, cela apporte également de nouveaux défis car le mécanisme interne des tables transactionnelles est plus complexe. J’ai rencontré plusieurs problèmes avec le système de compactions.

Que sont les compactions Hive ?

Avant de détailler les problèmes, vous devez comprendre ce que sont les compactions et pourquoi elles sont nécessaires avec Hive ACID.

Pour rendre les opérations ACID, chaque transaction (INSERT,UPDATE, DELETE) sur un bucket crée de nouveaux fichiers stockés dans des répertoires delta. Chaque partition de table contient un répertoire base et plusieurs répertoires delta :

hdfs dfs -ls -R /warehouse/tablespace/managed/hive/d.db/t
drwxr-xr-x   - hive hive     0 2016-06-09 17:03 /warehouse/tablespace/managed/hive/d.db/t/base_0000022
-rw-r--r--   1 hive hive   602 2016-06-09 17:03 /warehouse/tablespace/managed/hive/d.db/t/base_0000022/bucket_00000
drwxr-xr-x   - hive hive     0 2016-06-09 17:06 /warehouse/tablespace/managed/hive/d.db/t/delta_0000023_0000023_0000
-rw-r--r--   1 hive hive   611 2016-06-09 17:06 /warehouse/tablespace/managed/hive/d.db/t/delta_0000023_0000023_0000/bucket_00000
drwxr-xr-x   - hive hive     0 2016-06-09 17:07 /warehouse/tablespace/managed/hive/d.db/t/delta_0000024_0000024_0000
-rw-r--r--   1 hive hive   610 2016-06-09 17:07 /warehouse/tablespace/managed/hive/d.db/t/delta_0000024_0000024_0000/bucket_00000

Comme vous pouvez vous en douter, une partition contient autant de dossiers delta que de transactions (en réalité, les transactions sont regroupées et validées par lots). Cela peut rapidement devenir un problème en cas d’ingestion en continu, car Hive devra lire de plus en plus de fichiers à chaque requête.

Pour sauvegarder les performances de Hive, il existe 2 types de compactions :

• Les Compactions mineures pour fusionner les fichiers delta d’un bucket en un seul fichier delta ;
• Les compactions majeures pour fusionner tous les fichiers delta d’un bucket dans le fichier base existant de ce bucket.

Par défaut, Hive est censé déclencher automatiquement les compactions en se basant sur un ensemble de propriétés (voir la documentation officielle). Pourtant, selon mon expérience, cela ne fonctionne pas vraiment comme prévu…

Les compactions ne se déclenchent pas

Après avoir travaillé avec des tables transactionnelles en tant que destination d’un flux de données NiFi, j’ai constaté que les compactions ne fonctionnaient pas comme elles le devraient. Le comportement que j’ai observé est le suivant :

1. Une nouvelle partition est créée, elle n’a donc ni fichier de base ni de fichier delta ;
2. Hive inspecte cette partition pendant sa boucle de vérification. Vous pouvez voir quelles partitions sont contrôlées dans le fichier hivemetastore.log :

   cat /var/log/hive/hivemetastore.log | grep -P 'Checking to see if we should compact'

3. Comme il n’y a pas de fichier base, le seuil de construction du fichier base est atteint lors de la première vérification (car les fichiers delta représentent un pourcentage infini de la taille de la partition) ;
4. La compaction majeure est déclenchée et le fichier base est créé ;
5. Après cela, plus aucune vérification n’est faite sur cette partition…

Apparemment, je ne suis pas le seul à être confronté à ce problème [question Stack Overflow (edit : url indisponible), bien que je ne soit pas sûr que tout le monde y sera confronté, et je n’en ai pas trouvé la cause.

Après avoir tenté d’ajuster les paramètres hive.compactor.* sans succès, j’ai fini par écrire un workflow automatisé avec Apache Oozie pour déclencher “manuellement” les compactions majeures. Le workflow effectue les opérations suivantes au niveau d’une base de données toutes les 10 minutes :

1. Récupérer une liste des dossiers delta et base avec un script bash :

   # To get the delta list
   hdfs dfs -ls -R "/warehouse/tablespace/managed/hive/$target_db.db" \
   | grep 'drwx' | grep -P -o '[^ ]*(delta_\d{7}|base)_\d{7}_*\d{0,4}$' \

2. Analyser la liste et générer une requête de compaction pour chaque partition qui en a besoin. Je fais cela avec un script Python.
3. Publier les requêtes générées à Hive, ce que je fais via Beeline qui permet de soumettre un fichier de requête paramétré :

   -- Major compaction query example
   ALTER TABLE target_db.tbl1 PARTITION (col1 = 'value1') COMPACT 'major';

Voilà comment vous pouvez assurer les performances de Hive si vous rencontrez les mêmes problèmes avec les compactions. Si vous choisissez cette méthode, faites attention à :

• La queue YARN dans laquelle vous publiez vos requêtes de compaction. Pour éviter de pénaliser vos utilisateurs et vous assurer que les compactions ne seront pas retardées, je vous suggère de dédier une queue.
• La propriété yarn.scheduler.capacity[.queue].maximum-am-resource-percent, qui est de 20% par défaut. Chaque compaction lance 1 Application Master et 1 mapper. Vous pouvez donc augmenter le maximum-am-resource-percent, en particulier si vous dédiez une queue. Dans ce cas, cette propriété vous permet également de contrôler le nombre de compactions exécutées en parallèle.
• La mémoire utilisée par le compacteur. Si vous remarquez que certaines compactions prennent trop de temps, vous pouvez écraser la propriété compactor.mapreduce.map.memory.mb au niveau de la table ou lorsque vous requêtez la compaction :

  ALTER TABLE target_db.tbl1 PARTITION (col1 = 'value1')
  COMPACT 'major'
  WITH OVERWRITE TBLPROPERTIES ('compactor.mapreduce.map.memory.mb' = '3072');

• Le nombre maximal de fichiers delta compactés simultanément avec la propriété hive.compactor.max.num.delta. Cela contribue à uniformiser le temps pris par les compactions et à éviter les exceptions de type “OutOfMemory” :

  hive.compactor.max.num.delta = 100

• La propriété de table NO_AUTO_COMPACTION, qui est false par défaut. Si vous ne voulez pas que Hive interfère avec votre planification de compaction personnalisée, vous pouvez la définir à true sur des tables spécifiques :

  CREATE TABLE target_db.tbl1 (
    ...
  )
  TBLPROPERTIES ('NO_AUTO_COMPACTION' = 'true');

Échec d’impersonification du Cleaner

Si vous utilisez Apache Ranger pour gérer les autorisations dans Hive, la propriété hive.server2.enable.doAs est définie sur false afin que toute opération effectuée via le Hive Server soit exécutée en tant que hive. L’utilisateur hive est donc généralement propriétaire de tous les fichiers des tables managées.

Toutefois, lorsque vous utilisez HiveStreaming pour insérer des données dans des tables, vous ne communiquez pas avec le Hive Server, mais directement au serveur Hive Metastore ! C’est pourquoi vous devez faire attention aux informations d’identification de l’utilisateur qui appele l’API. Si hive n’est pas le propriétaire des partitions de table, vous pouvez rencontrer cette erreur dans le fichier hivemetastore.log :

2019-03-03T00:00:00,799 ERROR [pool-8-thread-188]: server.TThreadPoolServer (TThreadPoolServer.java:run(297)) - Error occurred during processing of message.
java.lang.RuntimeException: org.apache.thrift.transport.TTransportException: Peer indicated failure: GSS initiate failed
...
Caused by: org.apache.thrift.transport.TTransportException: Peer indicated failure: GSS initiate failed

Si vous activez le mode debug, vous obtiendrez l’origine de l’erreur :

2019-03-14T12:55:46,299 ERROR [Thread-19]: transport.TSaslTransport (TSaslTransport.java:open(315)) - SASL negotiation failure
javax.security.sasl.SaslException: GSS initiate failed
...
    at java.security.AccessController.doPrivileged(Native Method) [?:1.8.0_112]
    at javax.security.auth.Subject.doAs(Subject.java:422) [?:1.8.0_112]
...
    at org.apache.hadoop.hive.ql.txn.compactor.Cleaner.removeFiles(Cleaner.java:352) [hive-exec-3.1.0.3.0.1.0-187.jar:3.1.0.3.0.1.0-187]
    at org.apache.hadoop.hive.ql.txn.compactor.Cleaner.access$000(Cleaner.java:64) [hive-exec-3.1.0.3.0.1.0-187.jar:3.1.0.3.0.1.0-187]
    at org.apache.hadoop.hive.ql.txn.compactor.Cleaner$1.run(Cleaner.java:304) [hive-exec-3.1.0.3.0.1.0-187.jar:3.1.0.3.0.1.0-187]
    at java.security.AccessController.doPrivileged(Native Method) [?:1.8.0_112]
    at javax.security.auth.Subject.doAs(Subject.java:422) [?:1.8.0_112]
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1730) [hadoop-common-3.1.1.3.0.1.0-187.jar:?]
    at org.apache.hadoop.hive.ql.txn.compactor.Cleaner.clean(Cleaner.java:301) [hive-exec-3.1.0.3.0.1.0-187.jar:3.1.0.3.0.1.0-187]
    at org.apache.hadoop.hive.ql.txn.compactor.Cleaner.run(Cleaner.java:169) [hive-exec-3.1.0.3.0.1.0-187.jar:3.1.0.3.0.1.0-187]
Caused by: org.ietf.jgss.GSSException: No valid credentials provided (Mechanism level: Failed to find any Kerberos tgt)

Cette erreur signifie que le Cleaner, responsable de la suppression des fichiers delta obsolètes, ne peut pas emprunter l’identité de l’utilisateur à qui appartient la partition.

Le moyen le plus simple de résoudre ce problème est de rendre l’utilisateur hive propriétaire des partitions. Dans un environnement kerberisé, vous pouvez utiliser la keytab de l’utilisateur hive lors de l’interrogation de l’API HiveStreaming.

Requêter les données pendant les compactions

Bien que les propriétés ACID soit garanties lors des opérations INSERT,UPDATE et DELETE, ce que j’ai observé, c’est que le lancement d’une opération à l’échelle de la table pendant l’exécution de compactions peut entraîner des erreurs dues à la suppression de fichiers delta. Pour éviter ces problèmes de concurrence, je sépare l’insertion du requêtage avec l’architecture suivante :

• Une table est utilisée pour les opérations ACID, par exemple l’ingestion en temps réel depuis NiFi via HiveStreaming ;
• Une vue matérialisée est construite au-dessus de la table et est utilisée pour les requêtes.

Pour éviter le même problème que lors du requêtage, les compactions doivent être suspendues lors de la reconstruction de la vue matérialisée, qui, espérons-le, devrait être assez rapide grâce à la reconstruction incrémentale.

Conclusion

Dans cet article, nous avons passé en revue toutes les nouvelles fonctionnalités apportées par Hive 3. Elles sont nombreuses et attrayantes. Cependant, vous avez constaté qu’il n’est pas si facile de maintenir une instance de Hive 3. Cela devrait vous donner quelques avantages et inconvénients pour décider si vous devez ou non mettre à jour votre instance de Hive.

Astuces bonus

Voici quelques conseils qui pourraient vous faire gagner du temps.

Localisation du warehouse

Dans HDP 3, l’emplacement du warehouse Hive a changé. Vous pouvez trouver des tables managées dans HDFS dans chemin suivant :

/warehouse/tablespace/managed/hive/

Erreurs causées par les vues matérialisées

Pendant la maintenance des bases Hive, vous pouvez rencontrer cette erreur mystérieuse :

DROP TABLE my_table;
...
Error: Error while processing statement: FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. MetaException(message:Couldn't acquire the DB log notification lock because we reached the maximum # of retries: 5 retries. If this happens too often, then is recommended to increase the maximum number of retries on the hive.notification.sequence.lock.max.retries configuration :: Error executing SQL query "select "NEXT_EVENT_ID" from "NOTIFICATION_SEQUENCE" for update".) (state=08S01,code=1)

Même si ce message d’erreur vous donne une recommandation, ne la suivez pas trop vite ! Tout d’abord, allez vérifier les logs dans hivemetastore.log et vous trouverez certainement ceci :

2019-04-12T11:17:36,250 INFO  [pool-8-thread-174]: metastore.ObjectStore$RetryingExecutor (ObjectStore.java:run(9928)) - Attempting to acquire the DB log notification lock: 0 out of 5 retries
javax.jdo.JDODataStoreException: Error executing SQL query "select "NEXT_EVENT_ID" from "NOTIFICATION_SEQUENCE" for update".
...
Caused by: java.sql.BatchUpdateException: Batch entry 0 DELETE FROM "TBLS" WHERE "TBL_ID"=1852 was aborted: ERROR: update or delete on table "TBLS" violates foreign key constraint "MV_TABLES_USED_FK2" on table "MV_TABLES_USED"
  Detail: Key (TBL_ID)=(1852) is still referenced from table "MV_TABLES_USED".  Call getNextException to see other errors in the batch.
...
Caused by: org.postgresql.util.PSQLException: ERROR: update or delete on table "TBLS" violates foreign key constraint "MV_TABLES_USED_FK2" on table "MV_TABLES_USED"
  Detail: Key (TBL_ID)=(1852) is still referenced from table "MV_TABLES_USED".

Cela signifie qu’une vue matérialisée fait référence à la table que vous essayez de supprimer et que vous ne disposez pas des droits sur cette vue. Définissez les autorisations table/MV en conséquence et vous ne verrez plus ce message étrange.

Setup Hive

Les caractérisiques de l’instance Hive que je maintiens sont les suivantes :

• 2 Hive Metastores - 12 GB de RAM chacun
• 1 Hive Server 2 - 6 GB de RAM
• 1 Hive Server Interactive (LLAP) - 6 GB de RAM
• 3 daemons LLAP tournant sur 3 noeuds - 32 GB de RAM par daemon dont 18 GB pour le cache
• Le cluster est kerberizé avec Active Directory

Sources

• Apache Hive Wiki, https://cwiki.apache.org
• What’s new in Apache Hive presentation by Jason Dere, https://www.slideshare.net/Hadoop_Summit/whats-new-in-apache-hive-137569339
• Announcing HDP 3.0 – Faster, Smarter, Hybrid Data post on Hortonworks.com, http://web.archive.org/web/20190709182508/https://fr.hortonworks.com/blog/announcing-hdp-3-0-faster-smarter-hybrid-data/
• Cloudera documentation, https://docs.hortonworks.com/