Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# Connecteurs et utilitaires
Amazon EMR fournit un certain nombre de connecteurs et d'utilitaires permettant d'accéder à d'autres AWS services en tant que sources de données. Vous pouvez généralement accéder aux données de ces services au sein d'un programme. Par exemple, vous pouvez spécifier un flux Kinesis dans une requête Hive, un script Pig ou une MapReduce application, puis opérer sur ces données.
**Topics**
+ [Exportation, importation, interrogation et jointure de tables dans DynamoDB à l'aide d'Amazon EMR](EMRforDynamoDB.md)
+ [Kinesis](emr-kinesis.md)
+ [S3 DistCp (s3-dist-cp)](UsingEMR_s3distcp.md)
+ [Nettoyage après l'échec des DistCp tâches S3](#s3distcp-cleanup)
# Exportation, importation, interrogation et jointure de tables dans DynamoDB à l'aide d'Amazon EMR
**Note**
Le connecteur Amazon EMR-DynamoDB est open source sur. GitHub Pour de plus amples informations, veuillez consulter [https://github.com/awslabs/emr-dynamodb-connector](https://github.com/awslabs/emr-dynamodb-connector).
DynamoDB est un service de base de données NoSQL entièrement géré offrant des performances rapides et prévisibles avec une scalabilité simple. Les développeurs peuvent créer une table de base de données et développer sans limite son trafic de demandes ou son stockage. DynamoDB répartit automatiquement les données et le trafic de la table sur un nombre de serveurs approprié afin de gérer le volume de requêtes spécifié par le client et la quantité de données stockées, tout en assurant la cohérence et la rapidité des performances. Avec Amazon EMR et Hive, vous pouvez traiter rapidement et efficacement de grandes quantités de données, telles que les données stockées dans DynamoDB. Pour plus d'informations sur DynamoDB, consultez le [manuel du développeur Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/).
Apache Hive est une couche logicielle que vous pouvez utiliser pour interroger, mapper ou réduire les clusters à l'aide d'un langage de requête simplifié de type SQL, appelé HiveQL. Il s'exécute sur l'architecture Hadoop. Pour plus d'informations sur Hive et HiveQL, consultez le [Manuel du langage HiveQL](https://cwiki.apache.org/confluence/display/Hive/LanguageManual). Pour plus d'informations sur Hive et Amazon EMR, consultez [Apache Hive](emr-hive.md) .
Vous pouvez utiliser Amazon EMR avec une version personnalisée de Hive qui inclut une connectivité à DynamoDB pour effectuer des opérations sur les données stockées dans DynamoDB :
+ Chargement des données DynamoDB dans le système de fichier distribué Hadoop (HDFS) et son utilisation en tant qu'entrée dans un cluster Amazon EMR.
+ Interrogation de données DynamoDB en direct à l'aide d'instructions de type SQL (HiveQL).
+ Jointure de données stockées dans DynamoDB et exportation ou interrogation par rapport aux données jointes.
+ Exportation de données stockées dans DynamoDB vers Amazon S3.
+ Importation de données stockées dans Amazon S3 vers DynamoDB.
**Note**
Le connecteur Amazon EMR-DynamoDB ne prend pas en charge les clusters configurées pour utiliser l'[authentification Kerberos](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-kerberos.html).
Pour exécuter chacune des tâches suivantes, vous allez lancer un cluster Amazon EMR, spécifier l'emplacement des données dans DynamoDB et émettre des commandes Hive pour manipuler les données dans DynamoDB.
Il existe plusieurs manières de lancer un cluster Amazon EMR : vous pouvez utiliser la console Amazon EMR, l'interface de ligne de commande (CLI), ou vous pouvez programmer votre cluster à l'aide d'un AWS SDK ou de l'API Amazon EMR. Vous pouvez également choisir d'exécuter un cluster Hive de façon interactive ou à partir d'un script. Dans cette section, nous allons vous montrer comment lancer un cluster Hive interactif à partir de la console et de l'interface de ligne de commande Amazon EMR.
L'utilisation interactive de Hive est un excellent moyen de tester les performances des requêtes et de régler votre application. Une fois que vous avez défini un ensemble de commandes Hive qui s'exécuteront régulièrement, envisagez de créer un script Hive qu'Amazon EMR peut exécuter automatiquement.
**Avertissement**
Les opérations Amazon EMR de lecture et d'écriture sur une table DynamoDB desservent votre débit réservé établi, en augmentant potentiellement la fréquence des exceptions de débit réservé. Pour les demandes importantes, Amazon EMR implémente de nouvelles tentatives avec un backoff exponentiel pour gérer la charge des demandes sur la table DynamoDB. L'exécution simultanée de travaux Amazon EMR avec le reste du trafic peut entraîner un dépassement du niveau de débit réservé alloué. Vous pouvez contrôler cela en vérifiant la **ThrottleRequests**métrique sur Amazon CloudWatch. Si la charge de la demande est trop élevée, vous pouvez relancer le cluster et diminuer la valeur [Paramètre de pourcentage de lecture](EMR_Hive_Optimizing.md#ReadPercent) ou [Paramètre de pourcentage d'écriture](EMR_Hive_Optimizing.md#WritePercent) afin de limiter les opérations Amazon EMR. Pour de plus amples informations sur les paramètres de débit DynamoDB, veuillez consulter [Débit alloué](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithDDTables.html#ProvisionedThroughput).
Si une table est configurée pour le [mode à la demande](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.ReadWriteCapacityMode.html#HowItWorks.OnDemand), vous devez la remettre en mode provisionné avant d'exécuter une opération d'exportation ou d'importation. Les pipelines ont besoin d'un ratio de débit pour calculer les ressources à utiliser à partir d'une DynamoDBtable. Le mode à la demande supprime le débit provisionné. Pour provisionner la capacité de débit, vous pouvez utiliser les métriques Amazon CloudWatch Events pour évaluer le débit agrégé utilisé par une table.
**Topics**
+ [Configuration d'une table Hive pour exécuter des commandes Hive](EMR_Interactive_Hive.md)
+ [Exemples de commandes Hive pour l'exportation, l'importation et l'interrogation de données dans DynamoDB](EMR_Hive_Commands.md)
+ [Optimisation des performances pour les opérations Amazon EMR dans DynamoDB](EMR_Hive_Optimizing.md)
# Configuration d'une table Hive pour exécuter des commandes Hive
Apache Hive est une application d'entrepôt des données que vous pouvez utiliser pour interroger les données contenues dans les clusters Amazon EMR à l'aide d'un langage de type SQL. Pour plus d'informations sur Hive, consultez le site [http://hive.apache.org/](http://hive.apache.org/).
La procédure suivante suppose que vous avez déjà créé un cluster et spécifié une paire de clés Amazon EC2. Pour savoir comment commencer à créer des clusters, consultez la section [Premiers pas avec Amazon EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-gs) dans le *Guide de gestion Amazon EMR*.
## Configurer Hive pour utiliser MapReduce
Lorsque vous utilisez Hive sur Amazon EMR pour interroger les tables DynamoDB, des erreurs peuvent se produire si Hive utilise le moteur d'exécution par défaut, Tez. C'est pourquoi, lorsque vous créez un cluster avec Hive qui s'intègre à DynamoDB comme décrit dans cette section, nous vous recommandons d'utiliser une classification de configuration qui définit Hive à utiliser. MapReduce Pour de plus amples informations, veuillez consulter [Configuration des applications](emr-configure-apps.md).
L'extrait suivant montre la classification de configuration et les propriétés à utiliser pour définir MapReduce comme moteur d'exécution pour Hive :
```
[
{
"Classification": "hive-site",
"Properties": {
"hive.execution.engine": "mr"
}
}
]
```
**Pour exécuter les commandes Hive de façon interactive**
1. Connectez-vous au nœud maître. Pour plus d'informations, consultez [Connexion au nœud principal à l'aide de SSH](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html) dans le *Guide de gestion d'Amazon EMR*.
1. A l'invite de commande pour le nœud maître actuel, tapez `hive`.
Vous devez voir une invite Hive : `hive>`
1. Entrez une commande Hive qui mappe une table de l'application Hive aux données dans DynamoDB. Ce tableau agit comme référence pour les données stockées dans Amazon DynamoDB ; les données ne sont pas stockées localement dans Hive et les requêtes utilisant cette table s'exécutent sur les données en direct dans DynamoDB et consomment la capacité en lecture ou en écriture de la table chaque fois qu'une commande est exécutée. Si vous prévoyez d'exécuter plusieurs commandes Hive sur le même ensemble de données, pensez d'abord à l'exporter.
L'exemple suivant montre la syntaxe pour mapper une table Hive à une table DynamoDB.
```
CREATE EXTERNAL TABLE hive_tablename (hive_column1_name column1_datatype, hive_column2_name column2_datatype...)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodb_tablename",
"dynamodb.column.mapping" = "hive_column1_name:dynamodb_attribute1_name,hive_column2_name:dynamodb_attribute2_name...");
```
Lorsque vous créez une table dans Hive depuis DynamoDB, vous devez la créer en tant que table externe à l'aide du mot clé `EXTERNAL`. La différence entre les tables internes et externes est que les données des tables internes sont supprimées lorsqu'une table interne est supprimée. Il ne s'agit pas du comportement souhaité lorsque vous êtes connecté à Amazon DynamoDB, si bien que seules les tables externes sont prises en charge.
Par exemple, la commande Hive suivante crée une table nommée *hivetable1* dans Hive qui fait référence à la table DynamoDB nommée *dynamodbtable1*. La *table* DynamoDB dynamodbtable1 possède un schéma de clé primaire. hash-and-range L'élément de clé de hachage est `name` (type de chaîne), l'élément de clé de plage est `year` (type numérique) et chaque élément a une valeur d'attribut pour `holidays` (type string set).
```
CREATE EXTERNAL TABLE hivetable1 (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
```
La ligne 1 utilise l'instruction HiveQL `CREATE EXTERNAL TABLE`. Pour *hivetable1*, vous devez établir une colonne pour chaque paire nom-valeur d'attribut de la table DynamoDB et fournir le type de données. Ces valeurs ne sont pas sensibles à la casse et vous pouvez donner aux colonnes le nom de votre choix (à l'exception des mots réservés).
La ligne 2 utilise l'instruction `STORED BY`. La valeur de `STORED BY` est le nom de la classe qui gère la connexion entre Hive et DynamoDB. Cette valeur doit être définie sur `'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'`.
La ligne 3 utilise l'instruction `TBLPROPERTIES` pour associer « hivetable1 » à la table et au schéma appropriés dans DynamoDB. Fournissez à `TBLPROPERTIES` les valeurs des paramètres `dynamodb.table.name` et `dynamodb.column.mapping`. Ces valeurs *sont* sensibles à la casse.
**Note**
Tous les noms d'attribut DynamoDB de la table doivent avoir les colonnes correspondantes dans la table Hive. En fonction de votre version d'Amazon EMR, les scénarios suivants se produisent si le one-to-one mappage n'existe pas :
Sur Amazon EMR version 5.27.0 et versions ultérieures, le connecteur dispose de validations qui garantissent un mappage one-to-one entre les noms d'attributs DynamoDB et les colonnes de la table Hive. Une erreur se produira si le one-to-one mappage n'existe pas.
Sur Amazon EMR version 5.26.0 et antérieure, la table Hive ne contient pas de paire nom-valeur de DynamoDB. Si vous ne mappez pas les attributs de clé primaire DynamoDB, Hive génère une erreur. Si vous ne mappez pas un attribut de clé non primaire, aucune erreur n'est générée, mais vous ne voyez pas les données de la table Hive. Si les types de données ne correspondent pas, la valeur est null.
Ensuite, vous pouvez commencer à exécuter les opérations Hive sur *hivetable1*. Les requêtes exécutées sur *hivetable1* sont exécutées en interne sur la table DynamoDB *dynamodbtable1* de votre compte DynamoDB et consomment des unités de lecture ou d'écriture à chaque exécution.
Lorsque vous exécutez les requêtes Hive sur une table DynamoDB, vous devez vous assurer que vous avez alloué une quantité suffisante d'unités de capacité de lecture.
Par exemple, supposons que vous ayez provisionné 100 unités de capacité en lecture pour votre table DynamoDB. Vous pourrez ainsi effectuer 100 lectures, ou 409 600 octets, par seconde. Si cette table contient 20 Go de données (21 474 836 480 octets) et que votre requête Hive exécute une analyse complète de la table, vous pouvez estimer la durée d'exécution de la requête :
* 21 474 836 480 / 409 600 = 52 429 secondes = 14,56 heures *
Le seul moyen de diminuer le temps nécessaire consiste à ajuster les unités de capacité de lecture sur la table DynamoDB source. L'ajout de nœuds Amazon EMR supplémentaires n'y changera rien.
Dans la sortie Hive, le pourcentage d'achèvement est mis à jour lorsqu'un ou plusieurs processus de mappeurs sont terminés. Pour une table DynamoDB volumineuse avec un paramètre de faible capacité allouée en lecture, la sortie du pourcentage d'achèvement peut ne pas être mise à jour pendant longtemps. Dans le cas décrit ci-dessus, la tâche apparaît complète à 0 % pendant plusieurs heures. Pour un état plus détaillé de l'avancement de votre tâche, accédez à la console Amazon EMR ; vous serez en mesure d'afficher l'état des tâches de chaque mappeur et des statistiques de lecture des données. Vous pouvez également vous connecter à l'interface Hadoop sur le nœud maître et voir les statistiques Hadoop. Vous visualisez ainsi l'état des tâches de chaque carte et certaines statistiques de lecture des données. Pour plus d’informations, consultez les rubriques suivantes :
+ [Interfaces web hébergées sur le nœud principal](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html)
+ [Afficher les interfaces web Hadoop](https://docs.aws.amazon.com/emr/latest/ManagementGuide/UsingtheHadoopUserInterface.html)
Pour plus d'informations sur les exemples d'instructions HiveQL permettant d'exécuter des tâches telles que l'exportation ou l'importation de données depuis DynamoDB et la jointure de tables, consultez [Exemples de commandes Hive pour l'exportation, l'importation et l'interrogation de données dans DynamoDB](EMR_Hive_Commands.md).
**Pour annuler une demande Hive**
Lorsque vous exécutez une requête Hive, la réponse initiale du serveur inclut la commande pour annuler la demande. Pour annuler la demande à tout moment du processus, utilisez la **Kill Command (Commande kill)** de la réponse du serveur.
1. Entrez `Ctrl+C` pour quitter le client de ligne de commande.
1. À la commande shell, saisissez la **Commande kill** à partir de la réponse initiale du serveur à votre demande.
Vous pouvez également exécuter la commande suivante depuis la ligne de commande du nœud principal pour supprimer la tâche Hadoop. Elle contient l'identifiant de la tâche Hadoop et peut être récupérée depuis l'interface utilisateur de Hadoop. *job-id*
```
hadoop job -kill job-id
```
## Types de données pour Hive et DynamoDB
Le tableau suivant indique les types de données Hive disponibles, le type DynamoDB par défaut auquel ils correspondent et les autres types DynamoDB auxquels ils peuvent également mapper.
| Type Hive | Type DynamoDB par défaut | Autre(s) type(s) de DynamoDB |
| --- | --- | --- |
| chaîne | string (S) | |
| bigint ou double | number (N) | |
| binary | binary (B) | |
| boolean | boolean (BOOL) | |
| array | list (L) | number set (NS), string set (SS) ou binary set (BS) |
| map | article | map (M) |
| map | map (M) | |
| | null (NULL) | |
Si vous voulez écrire vos données Hive sous la forme d'un autre type DynamoDB correspondant, ou si vos données DynamoDB contiennent des valeurs d'attribut d'un autre type DynamoDB, vous pouvez spécifier la colonne et le type DynamoDB avec le paramètre `dynamodb.type.mapping`. L'exemple suivant montre la syntaxe permettant de spécifier un mappage d’un autre type.
```
CREATE EXTERNAL TABLE hive_tablename (hive_column1_name column1_datatype, hive_column2_name column2_datatype...)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodb_tablename",
"dynamodb.column.mapping" = "hive_column1_name:dynamodb_attribute1_name,hive_column2_name:dynamodb_attribute2_name...",
"dynamodb.type.mapping" = "hive_column1_name:dynamodb_attribute1_datatype");
```
Le paramètre de mappage de type est facultatif et ne doit être spécifié que pour les colonnes qui utilisent d'autres types.
Par exemple, la commande Hive suivante crée une table nommée `hivetable2` qui fait référence à la table DynamoDB `dynamodbtable2`. Elle est équivalent à `hivetable1`, sauf qu'elle mappe la colonne `col3` au type string set (SS).
```
CREATE EXTERNAL TABLE hivetable2 (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable2",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays",
"dynamodb.type.mapping" = "col3:SS");
```
Dans Hive, `hivetable1` et `hivetable2` sont identiques. Toutefois, lorsque les données de ces tables sont écrites dans leurs tables DynamoDB correspondantes, `dynamodbtable1` contient des listes, tandis que `dynamodbtable2` contient des ensembles de chaînes.
Si vous souhaitez écrire des valeurs `null` Hive en tant qu'attributs de type DynamoDB pour `null`, vous pouvez le faire avec le paramètre `dynamodb.null.serialization`. L'exemple suivant montre la syntaxe permettant de spécifier la sérialisation `null`.
```
CREATE EXTERNAL TABLE hive_tablename (hive_column1_name column1_datatype, hive_column2_name column2_datatype...)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodb_tablename",
"dynamodb.column.mapping" = "hive_column1_name:dynamodb_attribute1_name,hive_column2_name:dynamodb_attribute2_name...",
"dynamodb.null.serialization" = "true");
```
Le paramètre de sérialisation null est facultatif ; il est défini sur `false` s'il n'est pas spécifié. Notez que les attributs `null` DynamoDB sont lus en tant que valeurs `null` dans Hive, quel que soit le paramètre. Les collections Hive avec des valeurs `null` peuvent être écrites dans DynamoDB uniquement si le paramètre de sérialisation null est spécifié comme `true`. Sinon, une erreur Hive se produit.
Le type bigint de Hive est identique au type long de Java, et le type double de Hive est le même que le type double de Java en termes de précision. Cela signifie que si vous avez des données numériques stockées dans DynamoDB qui ont une précision supérieure à celle des types de données Hive, l'utilisation de Hive pour exporter, importer ou référencer les données DynamoDB peut entraîner une perte de précision ou un échec de la requête Hive.
Les exportations de type binaire de DynamoDB vers Amazon Simple Storage Service (Amazon S3) ou HDFS sont stockées sous la forme d'une chaîne codée en Base64. Si vous importez des données d'Amazon S3 ou HDFS dans le type binaire DynamoDB, elles doivent être encodées en tant que chaîne Base64.
## Options Hive
Vous pouvez définir les options Hive suivantes pour gérer le transfert de données hors d'Amazon DynamoDB. Ces options sont conservées uniquement pour la session Hive en cours. Si vous fermez l'invite de commande Hive et la rouvrez ultérieurement sur le cluster, ces paramètres sont rétablis à leurs valeurs par défaut.
| Options Hive | Description |
| --- | --- |
| dynamodb.throughput.read.percent | Définissez la vitesse des opérations de lecture pour maintenir votre débit DynamoDB dans la plage allouée pour votre table. La valeur est comprise entre `0.1` et `1.5`, valeurs incluses. La valeur de 0,5 est la vitesse de lecture par défaut, ce qui signifie que Hive essaie de consommer la moitié des lectures provisionnées dans le cadre des ressources de la table. L'augmentation de cette valeur au-dessus de 0,5 augmente le taux de demandes de lecture. Sa diminution sous 0,5 réduit le taux de demandes de lecture. Cette vitesse de lecture est approximative. La vitesse de lecture réelle dépend de facteurs comme l'existence d'une distribution uniforme des clés dans DynamoDB. Si vous trouvez que votre débit alloué est fréquemment dépassé par l'opération Hive, ou si le trafic de lecture en direct est trop limité, réduisez cette valeur sous `0.5`. Si vous avez suffisamment de capacité et voulez une opération Hive plus rapide, définissez cette valeur au-dessus de `0.5`. Vous pouvez également sursouscrire en la définissant sur 1,5 si vous pensez qu'il existe des opérations d'entrée/sortie inutilisées disponibles. |
| dynamodb.throughput.write.percent | Définissez la vitesse des opérations d'écriture pour maintenir votre débit DynamoDB dans la plage allouée pour votre table. La valeur est comprise entre `0.1` et `1.5`, valeurs incluses. La valeur de 0,5 est la vitesse d'écriture par défaut, ce qui signifie que Hive essaie d'utiliser la moitié des écritures provisionnées dans le cadre des ressources de la table. L'augmentation de cette valeur au-dessus de 0,5 augmente le taux de demandes d'écriture. Sa diminution sous 0,5 réduit le taux de demandes d'écriture. Cette vitesse d'écriture est approximative. La vitesse d'écriture réelle dépend de facteurs comme l'existence d'une distribution uniforme des clés dans DynamoDB. Si vous trouvez que votre débit alloué est fréquemment dépassé par l'opération Hive ou si le trafic d'écriture en direct est trop limité, réduisez cette valeur sous `0.5`. Si vous avez suffisamment de capacité et voulez une opération Hive plus rapide, définissez cette valeur au-dessus de `0.5`. Vous pouvez également sursouscrire en la définissant jusqu'à 1,5 si vous pensez qu'il existe des opérations d'entrée/sortie inutilisées disponibles ou qu'il s'agit du chargement des données initiales sur la table et qu'il n'y pas encore de trafic en direct. |
| dynamodb.endpoint | Spécifiez le point de terminaison pour le service DynamoDB. Pour plus d'information sur les points de terminaison DynamoDB disponibles, consultez [Régions et points de terminaison](https://docs.aws.amazon.com/general/latest/gr/rande.html#ddb_region). |
| dynamodb.max.map.tasks | Spécifiez le nombre maximal de tâches de mappage lors de la lecture des données à partir de DynamoDB. Cette valeur doit être supérieure ou égale à 1. |
| dynamodb.retry.duration | Spécifiez le nombre de minutes à utiliser comme délai d'expiration pour répéter les commandes Hive. Cette valeur doit être un entier supérieur ou égal à 0. Le délai d'expiration par défaut est de deux minutes. |
Ces options sont définies à l'aide de la commande `SET`, comme illustré dans l'exemple suivant.
```
SET dynamodb.throughput.read.percent=1.0;
INSERT OVERWRITE TABLE s3_export SELECT *
FROM hiveTableName;
```
# Exemples de commandes Hive pour l'exportation, l'importation et l'interrogation de données dans DynamoDB
Les exemples suivants utilisent les commandes Hive afin d'effectuer des opérations telles que l'exportation de données vers Amazon S3 ou HDFS, l'importation de données vers DynamoDB, la jointure de tables, l'interrogation de tables, et plus encore.
Les opérations sur une table Hive référencent les données stockées dans DynamoDB. Les commandes Hive sont soumises aux paramètres de débit alloué de la table DynamoDB et les données extraites incluent les données écrites dans la table DynamoDB au moment où la demande d'opération Hive est traitée par DynamoDB. Si le processus de récupération des données prend du temps, certaines données retournées par la commande Hive peuvent avoir été mises à jour dans DynamoDB depuis que la commande Hive a commencé.
Les commandes Hive `DROP TABLE` et `CREATE TABLE` agissent uniquement sur les tables locales de Hive et ne créent ni ne suppriment de tables dans DynamoDB. Si votre requête Hive fait référence à une table dans DynamoDB, cette table doit déjà exister avant que vous exécutiez la requête. Pour plus d'informations sur la création et la suppression de tables dans DynamoDB, consultez [Utilisation de tables dans DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.html) dans le *Guide du développeur Amazon DynamoDB*.
**Note**
Lorsque vous mappez une table Hive à un emplacement dans Amazon S3, ne la mappez pas au chemin racine du bucket, s3://amzn-s3-demo-bucket, car cela peut provoquer des erreurs lorsque Hive écrit les données dans Amazon S3. Mappez plutôt la table à un sous-chemin du bucket, s3://amzn-s3-demo-bucket/mypath.
## Exportation de données à partir de DynamoDB
Vous pouvez utiliser Hive pour exporter des données à partir de DynamoDB.
**Pour exporter une table DynamoDB vers un compartiment Amazon S3**
+ Créez une table Hive qui référence les données stockées dans DynamoDB. Ensuite, vous pouvez appeler la commande INSERT OVERWRITE pour écrire les données dans un répertoire externe. Dans l'exemple suivant, *s3://amzn-s3-demo-bucket/path/subpath/* il s'agit d'un chemin valide dans Amazon S3. Ajustez les colonnes et les types de données dans la commande CREATE pour faire correspondre les valeurs dans votre DynamoDB. Vous pouvez l'utiliser pour créer une archive de vos données DynamoDB dans Amazon S3.
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. INSERT OVERWRITE DIRECTORY 's3://amzn-s3-demo-bucket/path/subpath/' SELECT *
7. FROM hiveTableName;
```
**Pour exporter une table DynamoDB vers un compartiment Amazon S3 à l'aide de la compartimentation.**
+ Créez une table externe qui fait référence à un emplacement dans Amazon S3. Cela est présenté ci-dessous en tant que s3\$1export. Pendant l'appel CREATE, spécifiez le formatage des lignes de la table. Puis, lorsque vous utilisez INSERT OVERWRITE pour exporter les données de DynamoDB vers s3\$1export, les données sont écrites au format spécifié. Dans l'exemple suivant, les données sont écrites sous forme de valeurs séparées par des virgules (CSV).
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. CREATE EXTERNAL TABLE s3_export(a_col string, b_col bigint, c_col array)
7. ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
8. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
9.
10. INSERT OVERWRITE TABLE s3_export SELECT *
11. FROM hiveTableName;
```
**Pour exporter une table DynamoDB vers un compartiment Amazon S3 sans spécifier de mappage de colonne**
+ Créez une table Hive qui référence les données stockées dans DynamoDB. Cet exemple est similaire au précédent, sauf que vous ne spécifiez pas un mappage de colonne. La table doit avoir exactement une colonne de type `map`. Si vous créez ensuite une table `EXTERNAL` dans Amazon S3, vous pouvez appeler la commande `INSERT OVERWRITE` pour écrire les données de DynamoDB vers Amazon S3. Vous pouvez l'utiliser pour créer une archive de vos données DynamoDB dans Amazon S3. Comme il n'y a aucun mappage de colonnes, vous ne pouvez pas interroger les tables qui sont exportées de cette façon. L'exportation de données sans spécifier de mappage de colonne est disponible dans Hive 0.8.1.5 (ou version ultérieure), qui est prise en charge sur l'AMI Amazon EMR 2.2.*x* et les versions ultérieures.
```
1. CREATE EXTERNAL TABLE hiveTableName (item map)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1");
4.
5. CREATE EXTERNAL TABLE s3TableName (item map)
6. ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
7. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
8.
9. INSERT OVERWRITE TABLE s3TableName SELECT *
10. FROM hiveTableName;
```
**Pour exporter une table DynamoDB vers un compartiment Amazon S3 en utilisant la compartimentation des données**
+ Hive fournit plusieurs codecs de compression, que vous pouvez définir durant votre session Hive. Les données exportées sont ainsi compressées au format spécifié. L'exemple suivant compresse les fichiers exportés à l'aide de l'algorithme Lempel-Ziv-Oberhumer (LZO).
```
1. SET hive.exec.compress.output=true;
2. SET io.seqfile.compression.type=BLOCK;
3. SET mapred.output.compression.codec = com.hadoop.compression.lzo.LzopCodec;
4.
5. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
6. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
7. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
8. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
9.
10. CREATE EXTERNAL TABLE lzo_compression_table (line STRING)
11. ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
12. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
13.
14. INSERT OVERWRITE TABLE lzo_compression_table SELECT *
15. FROM hiveTableName;
```
Les codecs de compression disponibles sont les suivants :
+ org.apache.hadoop.io.compress. GzipCodec
+ org.apache.hadoop.io.compress. DefaultCodec
+ com.hadoop.compression.lzo. LzoCodec
+ com.hadoop.compression.lzo. LzopCodec
+ org.apache.hadoop.io.compress. BZip2Codec
+ org.apache.hadoop.io.compress. SnappyCodec
**Pour exporter une table DynamoDB vers HDFS**
+ Utilisez la commande Hive suivante, où se *hdfs:///directoryName* trouve un chemin HDFS valide et *hiveTableName* une table dans Hive qui fait référence à DynamoDB. Cette opération d'exportation est plus rapide que l'exportation d'une table DynamoDB vers Amazon S3, car Hive 0.7.1.1 utilise HDFS comme étape intermédiaire lors de l'exportation des données vers Amazon S3. L'exemple suivant montre aussi comment définir `dynamodb.throughput.read.percent` sur 1.0 afin d'augmenter le taux de demande de lecture.
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. SET dynamodb.throughput.read.percent=1.0;
7.
8. INSERT OVERWRITE DIRECTORY 'hdfs:///directoryName' SELECT * FROM hiveTableName;
```
Vous pouvez également exporter les données vers HDFS à l'aide de la mise en forme et de la compression comme indiqué ci-dessus pour l'exportation vers Amazon S3. Pour ce faire, remplacez simplement le répertoire Amazon S3 dans les exemples ci-dessus par un répertoire HDFS.
**Pour lire les données de caractères UTF-8 non affichables dans Hive**
+ Vous pouvez lire et écrire des données de caractères UTF-8 non affichables avec Hive en utilisant la clause `STORED AS SEQUENCEFILE` lorsque vous créez la table. A SequenceFile est le format de fichier binaire Hadoop ; vous devez utiliser Hadoop pour lire ce fichier. L’exemple suivant montre comment exporter des données de DynamoDB vers Amazon S3. Vous pouvez utiliser cette fonctionnalité pour gérer les caractères codés UTF-8 non affichables.
```
1. CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
2. STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
3. TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
4. "dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
5.
6. CREATE EXTERNAL TABLE s3_export(a_col string, b_col bigint, c_col array)
7. STORED AS SEQUENCEFILE
8. LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
9.
10. INSERT OVERWRITE TABLE s3_export SELECT *
11. FROM hiveTableName;
```
## Importation de données dans DynamoDB
Lorsque vous écrivez des données sur DynamoDB à l'aide de Hive, vous devez vous assurer que le nombre d'unités de capacité d'écriture est supérieur au nombre de mappeurs dans le cluster. Par exemple, les clusters qui s'exécutent sur les instances EC2 m1.xlarge produisent 8 mappeurs par instance. Dans le cas d'un cluster qui possède 10 instances, cela signifie un total de 80 mappeurs. Si vos unités de capacité en écriture ne sont pas supérieures au nombre de mappeurs dans le cluster, l'opération d'écriture Hive peut consommer la totalité du débit d'écriture ou tenter d'utiliser plus de débit qu'il n'en est alloué. Pour plus d'informations sur le nombre de mappeurs émis par chaque type d'instance EC2, consultez [Configuration de Hadoop](emr-hadoop-config.md).
Le nombre de mappeurs dans Hadoop est contrôlé par les divisions en entrée. S'il y a trop peu de divisions, votre commande d'écriture peut ne pas être en mesure d'utiliser tout le débit d'écriture disponible.
Si un élément avec la même clé existe dans la table DynamoDB cible, il est remplacé. Si aucun élément avec la clé n'existe dans la table DynamoDB cible, l'élément est inséré.
**Pour importer une table d'Amazon S3 vers DynamoDB**
+ Vous pouvez utiliser Amazon EMR (Amazon EMR) et Hive pour écrire des données d'Amazon S3 vers DynamoDB.
```
CREATE EXTERNAL TABLE s3_import(a_col string, b_col bigint, c_col array)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
INSERT OVERWRITE TABLE hiveTableName SELECT * FROM s3_import;
```
**Pour importer une table depuis un compartiment Amazon S3 vers DynamoDB sans spécifier de mappage de colonne**
+ Créez une table `EXTERNAL` qui référence les données stockées dans Amazon S3 et précédemment exportées depuis DynamoDB. Avant de les importer, vérifiez que la table existe dans DynamoDB et qu'elle possède le même schéma de clés que la table DynamoDB précédemment exportée. En outre, la table doit avoir une seule colonne de type `map`. Si vous créez ensuite une table Hive liée à DynamoDB, vous pouvez appeler la commande `INSERT OVERWRITE` pour écrire les données depuis Amazon S3 vers DynamoDB. Comme il n'existe aucun mappage de colonne, vous ne pouvez pas interroger les tables qui sont importées de cette façon. L'importation de données sans spécifier un mappage de colonne est disponible dans Hive 0.8.1.5 ou version ultérieure, qui est pris en charge sur Amazon EMR AMI 2.2.3.x et version ultérieure.
```
CREATE EXTERNAL TABLE s3TableName (item map)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
CREATE EXTERNAL TABLE hiveTableName (item map)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1");
INSERT OVERWRITE TABLE hiveTableName SELECT *
FROM s3TableName;
```
**Pour importer une table de HDFS vers DynamoDB**
+ Vous pouvez utiliser Amazon EMR et Hive pour écrire des données de HDFS vers DynamoDB.
```
CREATE EXTERNAL TABLE hdfs_import(a_col string, b_col bigint, c_col array)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 'hdfs:///directoryName';
CREATE EXTERNAL TABLE hiveTableName (col1 string, col2 bigint, col3 array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "dynamodbtable1",
"dynamodb.column.mapping" = "col1:name,col2:year,col3:holidays");
INSERT OVERWRITE TABLE hiveTableName SELECT * FROM hdfs_import;
```
## Interrogation de données dans DynamoDB
Les exemples suivants illustrent les différentes façons dont vous pouvez utiliser Amazon EMR pour interroger les données stockées dans DynamoDB.
**Pour trouver la plus grande valeur pour une colonne mappée (`max`)**
+ Utilisez les commandes de Hive comme suit. Dans la première commande, l'instruction CREATE crée une table Hive qui référence les données stockées dans DynamoDB. L'instruction SELECT utilise ensuite cette table pour interroger les données stockées dans DynamoDB. L'exemple suivant recherche la plus grande commande passée par un client donné.
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
SELECT max(total_cost) from hive_purchases where customerId = 717;
```
**Pour regrouper les données à l'aide de la clause `GROUP BY`**
+ Vous pouvez utiliser la clause `GROUP BY` pour collecter les données de plusieurs enregistrements. Cette option est souvent utilisée avec une fonction d'agrégation comme somme, décompte, min ou max. L'exemple suivant retourne une liste des commandes les plus importantes des clients ayant passé plus de trois commandes.
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
SELECT customerId, max(total_cost) from hive_purchases GROUP BY customerId HAVING count(*) > 3;
```
**Pour joindre deux tables DynamoDB**
+ L'exemple suivant mappe deux tables Hive avec des données stockées dans DynamoDB. Il appelle ensuite une jointure entre ces deux tables. La jointure est calculée sur le cluster, puis renvoyée. La jointure n’a pas lieu dans DynamoDB. Cet exemple renvoie une liste de clients et de leurs achats pour ceux ayant passé plus de deux commandes.
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
CREATE EXTERNAL TABLE hive_customers(customerId bigint, customerName string, customerAddress array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Customers",
"dynamodb.column.mapping" = "customerId:CustomerId,customerName:Name,customerAddress:Address");
Select c.customerId, c.customerName, count(*) as count from hive_customers c
JOIN hive_purchases p ON c.customerId=p.customerId
GROUP BY c.customerId, c.customerName HAVING count > 2;
```
**Pour joindre deux tables de différentes sources**
+ Dans l'exemple suivant, Customer\$1S3 est une table Hive qui charge un fichier CSV stocké dans Amazon S3 et hive\$1purchases est une table qui fait référence aux données de DynamoDB. L'exemple suivant joint les données client stockées comme fichier CSV dans Amazon S3 et les données relatives aux commandes stockées dans DynamoDB pour renvoyer un ensemble de données représentant les commandes passées par les clients ayant « Miller » dans leur nom.
```
CREATE EXTERNAL TABLE hive_purchases(customerId bigint, total_cost double, items_purchased array)
STORED BY 'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'
TBLPROPERTIES ("dynamodb.table.name" = "Purchases",
"dynamodb.column.mapping" = "customerId:CustomerId,total_cost:Cost,items_purchased:Items");
CREATE EXTERNAL TABLE Customer_S3(customerId bigint, customerName string, customerAddress array)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
LOCATION 's3://amzn-s3-demo-bucket/path/subpath/';
Select c.customerId, c.customerName, c.customerAddress from
Customer_S3 c
JOIN hive_purchases p
ON c.customerid=p.customerid
where c.customerName like '%Miller%';
```
**Note**
Dans les exemples précédents, les instructions CREATE TABLE ont été incluses dans chaque exemple pour des raisons de clarté et d'exhaustivité. Lorsque vous exécutez plusieurs requêtes ou opérations d'exportation sur une table Hive donnée, vous ne devez créer la table qu'une seule fois, au début de la session Hive.
# Optimisation des performances pour les opérations Amazon EMR dans DynamoDB
Les opérations Amazon EMR sur une table DynamoDB sont considérées comme des opérations de lecture, et sont soumises aux paramètres de débit approvisionnés de la table. Amazon EMR implémente sa propre logique pour l'équilibrage de la charge sur votre table DynamoDB, afin de minimiser la possibilité de dépassement du débit approvisionné. À la fin de chaque requête Hive, Amazon EMR renvoie des informations sur le cluster utilisé pour traiter la requête, y compris le nombre de fois où votre débit alloué a été dépassé. Vous pouvez utiliser ces informations, ainsi que CloudWatch les mesures relatives à votre débit DynamoDB, pour mieux gérer la charge sur votre table DynamoDB lors des demandes suivantes.
Les facteurs suivants influencent les performances des requêtes Hive lorsque vous travaillez avec des tables DynamoDB.
## Unités de capacité de lecture allouées
Lorsque vous exécutez les requêtes Hive sur une table DynamoDB, vous devez vous assurer que vous avez alloué une quantité suffisante d'unités de capacité de lecture.
Par exemple, supposons que vous ayez provisionné 100 unités de capacité en lecture pour votre table DynamoDB. Vous pourrez ainsi effectuer 100 lectures, ou 409 600 octets, par seconde. Si cette table contient 20 Go de données (21 474 836 480 octets) et que votre requête Hive exécute une analyse complète de la table, vous pouvez estimer la durée d'exécution de la requête :
* 21 474 836 480 / 409 600 = 52 429 secondes = 14,56 heures *
Le seul moyen de diminuer le temps nécessaire consiste à ajuster les unités de capacité de lecture sur la table DynamoDB source. L'ajout de nœuds supplémentaires au cluster Amazon EMR ne sera d'aucune utilité.
Dans la sortie Hive, le pourcentage d'achèvement est mis à jour lorsqu'un ou plusieurs processus de mappeurs sont terminés. Pour une table DynamoDB volumineuse avec un paramètre de faible capacité allouée en lecture, la sortie du pourcentage d'achèvement peut ne pas être mise à jour pendant longtemps. Dans le cas décrit ci-dessus, la tâche apparaît complète à 0 % pendant plusieurs heures. Pour un état plus détaillé de l'avancement de votre tâche, accédez à la console Amazon EMR ; vous serez en mesure d'afficher l'état des tâches de chaque mappeur et des statistiques de lecture des données.
Vous pouvez également vous connecter à l'interface Hadoop sur le nœud maître et voir les statistiques Hadoop. Vous visualisez ainsi l'état des tâches de chaque carte et certaines statistiques de lecture des données. Pour plus d'informations, consultez [Interfaces web hébergées sur le nœud principal](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html) dans le *Guide de gestion d'Amazon EMR*.
## Paramètre de pourcentage de lecture
Par défaut, Amazon EMR gère la charge de requête sur votre table DynamoDB selon votre débit alloué actuel. Cependant, quand Amazon EMR renvoie des informations sur votre tâche avec un nombre élevé de réponses de dépassement du débit alloué, vous pouvez ajuster la vitesse de lecture par défaut à l'aide du paramètre `dynamodb.throughput.read.percent` lorsque vous configurez la table Hive. Pour plus d'informations sur la configuration du paramètre de pourcentage de lecture, consultez [Options Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Paramètre de pourcentage d'écriture
Par défaut, Amazon EMR gère la charge de requête sur votre table DynamoDB selon votre débit alloué actuel. Cependant, quand Amazon EMR renvoie des informations sur votre tâche avec un nombre élevé de réponses de dépassement du débit alloué, vous pouvez ajuster la vitesse d'écriture par défaut à l'aide du paramètre `dynamodb.throughput.write.percent` lorsque vous configurez la table Hive. Pour plus d'informations sur la configuration du paramètre de pourcentage d'écriture, consultez [Options Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Paramètre de durée de nouvelle tentative
Par défaut, Amazon EMR exécute à nouveau une requête Hive si elle n'a pas renvoyé de résultat dans les deux minutes, l'intervalle de nouvelle tentative par défaut. Vous pouvez ajuster cet intervalle en définissant le paramètre `dynamodb.retry.duration` lorsque vous exécutez une requête Hive. Pour plus d'informations sur la configuration du paramètre de pourcentage d'écriture, consultez [Options Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Nombre de tâches de mappage
Les démons de mappeur que Hadoop lance pour traiter vos requêtes afin d'exporter et d'interroger des données stockées dans DynamoDB sont limités à une vitesse de lecture maximale de 1 Mio par seconde afin de limiter la capacité de lecture utilisée. Si vous avez un débit alloué supplémentaire disponible sur DynamoDB, vous pouvez améliorer les performances de l'exportation Hive et des opérations d'interrogation en augmentant le nombre de démons de mappeur. Pour ce faire, vous pouvez augmenter le nombre d'instances EC2 dans votre cluster *ou* accroître le nombre de démons de mappeur s'exécutant sur chaque instance EC2.
Vous pouvez augmenter le nombre d'instances EC2 dans un cluster en arrêtant le cluster actuel et en le relançant avec un plus grand nombre d'instances EC2. Vous spécifiez le nombre d'instances EC2 dans la boîte de dialogue **Configurer les instances EC2** si vous lancez le cluster à partir de la console Amazon EMR, ou avec l'option `‑‑num-instances` si vous lancez le cluster à partir de la CLI.
Le nombre de tâches de mappage exécutées sur une instance dépend du type d'instance EC2. Pour plus d'informations sur les types d'instance EC2 pris en charge et le nombre de mappeurs que chacun fournit, consultez [Configuration de la tâche](emr-hadoop-task-config.md). Là, vous trouverez une section « Configuration des tâches » pour chacune des configurations prises en charge.
Un autre moyen d'augmenter le nombre de démons de mappeur consiste à modifier le paramètre de configuration `mapreduce.tasktracker.map.tasks.maximum` de Hadoop sur une valeur plus élevée. Cela a l'avantage de vous donner un plus grand nombre de mappeur sans augmenter le nombre ou la taille des instances EC2, ce qui vous permet d'économiser de l'argent. Un inconvénient est qu'une définition trop élevée de cette valeur peut entraîner une saturation de la mémoire des instances EC2 dans votre cluster. Pour définir `mapreduce.tasktracker.map.tasks.maximum`, lancez le cluster et spécifiez une valeur pour `mapreduce.tasktracker.map.tasks.maximum` en tant que propriété de la classification de configuration mapred-site. Voici un exemple : Pour de plus amples informations, veuillez consulter [Configuration des applications](emr-configure-apps.md).
```
{
"configurations": [
{
"classification": "mapred-site",
"properties": {
"mapred.tasktracker.map.tasks.maximum": "10"
}
}
]
}
```
## Demandes de données en parallèle
Plusieurs demandes de données, à partir de plusieurs utilisateurs ou de plusieurs applications vers une seule table, peuvent diminuer la vitesse de lecture allouée et ralentir les performances.
## Durée du processus
La cohérence des données dans DynamoDB dépend de l’ordre des opérations de lecture et d’écriture sur chaque nœud. Quand une requête Hive est en cours, une autre application peut charger de nouvelles données dans la table DynamoDB, voire modifier ou supprimer des données existantes. Dans ce cas, les résultats de la requête Hive peuvent ne pas tenir compte des modifications apportées aux données pendant l’exécution de la requête.
## Eviter le dépassement de débit
Lors de l'exécution de requêtes Hive sur DynamoDB, veillez à ne pas dépasser votre débit alloué, vous risquez sinon de réduire la capacité nécessaire des appels à `DynamoDB::Get` de votre application. Pour éviter que cela ne se produise, vous devez surveiller régulièrement le volume de lecture et la limitation des appels aux applications `DynamoDB::Get` en consultant les journaux et les métriques de surveillance sur Amazon. CloudWatch
## Durée de la demande
La planification des requêtes Hive qui accèdent à une table DynamoDB à un moment où la demande sur celle-ci est inférieure a pour effet d’améliorer les performances. Par exemple, si la plupart des utilisateurs de votre application vivent à San Francisco, vous pouvez choisir d'exporter les données chaque jour à 4h00 HNP, lorsque la majorité des utilisateurs dort, sans mettre à jour les enregistrements de votre base de données DynamoDB.
## Tables basées sur le temps
Si les données sont organisées en tant que série de tables DynamoDB basées sur le temps, telles qu'une table par jour, vous pouvez exporter les données lorsque la table n'est plus active. Vous pouvez utiliser cette technique pour sauvegarder les données sur Amazon S3 en continu.
## Données archivées
Si vous prévoyez d'exécuter plusieurs requêtes Hive sur les données stockées dans DynamoDB et que votre application peut tolérer les données archivées, il se peut que vous souhaitiez exporter les données vers HDFS ou Amazon S3, et exécuter les requêtes Hive sur une copie des données au lieu de DynamoDB. Cela permet de conserver vos opérations de lecture et votre débit alloué.
# Kinesis
Les clusters Amazon EMR peuvent lire et traiter directement les flux Amazon Kinesis à l'aide d'outils courants de l'écosystème Hadoop tels que Hive, Pig MapReduce, l'API de streaming Hadoop et Cascading. Vous pouvez également joindre les données en temps réel d'Amazon Kinesis aux données existantes sur Amazon S3, Amazon DynamoDB et HDFS dans un cluster en cours d'exécution. Vous pouvez charger directement les données d'Amazon EMR sur Amazon S3 ou DynamoDB pour les activités de post-traitement. Pour plus d’informations sur les caractéristiques principales du service Amazon Kinesis et ses tarifs, voir la page [Amazon Kinesis](https://aws.amazon.com//kinesis).
## Que puis-je faire avec l'intégration d'Amazon EMR et Amazon Kinesis ?
L'intégration entre Amazon EMR et Amazon Kinesis facilite considérablement certains scénarios, par exemple :
+ **Analyse de journaux de streaming** – Vous pouvez analyser les journaux web de vos flux de diffusion continus afin de générer la liste des 10 principaux types d'erreurs par région, par navigateur et par domaine d'accès à intervalles de quelques minutes.
+ **Engagement client** – Vous pouvez écrire des requêtes associant des données de parcours provenant d'Amazon Kinesis à des informations sur une campagne publicitaire stockées dans une table DynamoDB, dans le but d'identifier les catégories de publicité les plus efficaces parmi celles affichées sur des sites web donnés.
+ **Requêtes interactives ad-hoc** – Vous pouvez charger périodiquement des données provenant de flux Amazon Kinesis dans HDFS et les mettre à disposition sous forme d'une table Impala locale permettant des requêtes analytiques, rapides et interactives.
## Analyse de points de contrôle des flux Amazon Kinesis
Les utilisateurs peuvent exécuter des analyses régulières par lots de flux Amazon Kinesis dans ce que l'on appelle des *itérations*. Étant donné que les enregistrements de données de flux Amazon Kinesis sont récupérés à l'aide d'un numéro de séquence, des limites d'itérations sont définies par des numéros de début et de fin de séquences qu'Amazon EMR stocke dans une table DynamoDB. Par exemple, quand `iteration0` se termine, elle stocke le numéro de fin de séquence dans DynamoDB afin que lorsque la tâche `iteration1` commence, elle puisse extraire les données suivantes du flux. Ce mappage d'itérations de données de flux s'appelle un *point de contrôle*. Pour plus d'informations, consultez [Connecteur Kinesis](https://aws.amazon.com/elasticmapreduce/faqs/#kinesis-connector).
Si une itération a été vérifiée et que la tâche a échoué lors du traitement d'une itération, Amazon EMR tente de retraiter les enregistrements de cette itération.
Les points de contrôle sont une fonctionnalité qui vous permet de :
+ Commencer le traitement de données après un numéro de séquence traité par une requête précédente qui s'est exécutée sur le même flux de données et nom logique
+ Retraiter le même lot de données à partir de Kinesis qui a été traité par une requête précédente
Pour activer les points de contrôle, définissez le paramètre `kinesis.checkpoint.enabled` sur `true` dans vos scripts. En outre, configurez les paramètres suivants :
| Paramètre de configuration | Description |
| --- | --- |
| kinesis.checkpoint.metastore.table.name | Nom de la table DynamoDB où seront stockées les informations de point de contrôle |
| kinesis.checkpoint.metastore.hash.key.name | Nom de la clé de hachage pour la table DynamoDB |
| kinesis.checkpoint.metastore.hash.range.name | Nom de la clé de plage pour la table DynamoDB |
| kinesis.checkpoint.logical.name | Un nom logique pour le traitement actuel |
| kinesis.checkpoint.iteration.no | Numéro de l'itération pour le traitement associé au nom logique |
| kinesis.rerun.iteration.without.wait | Valeur booléenne qui indique si une itération ayant échoué peut être exécutée à nouveau sans attente l'expiration ; la valeur par défaut est false |
### Recommandations d'IOPS provisionnés pour les tables Amazon DynamoDB
Le connecteur Amazon EMR pour Amazon Kinesis utilise la base de données DynamoDB comme support pour les métadonnées de pointage. Vous devez créer une table dans DynamoDB avant de consommer des données dans un flux Amazon Kinesis avec un cluster Amazon EMR à intervalles contrôlés. La table doit se trouver dans la même région que votre cluster Amazon EMR. Voici des recommandations générales pour le nombre d'IOPS que vous devez mettre en service pour vos tables DynamoDB. `j` est le nombre maximum de tâches Hadoop (avec une combinaison de nombre d'itérations\$1nom logique différente) pouvant s'exécuter simultanément et `s` est le nombre maximum de partitions que toute tâche traitera :
Pour **Read Capacity Units (Unités de capacité en lecture)** : `j`\$1`s`/`5`
Pour **Write Capacity Units (Unités de capacité en écriture)** : `j`\$1`s`
## Considérations sur les performances
Le débit de la partition Amazon Kinesis est directement proportionnel à la taille de l'instance des nœuds dans les clusters Amazon EMR et à la taille de l'enregistrement dans le flux. Nous vous recommandons d'utiliser des instances m5.xlarge ou plus importantes sur les nœuds maîtres et principaux.
## Planifiez une analyse Amazon Kinesis avec Amazon EMR
Lorsque vous analysez des données sur un flux Amazon Kinesis actif, limité par des délais d'attente et une durée maximale pour l'itération, il est important que vous exécutiez l'analyse fréquemment pour collecter régulièrement des détails à partir du flux. Il existe plusieurs façons d'exécuter ces scripts et requêtes à intervalles réguliers ; nous vous conseillons d'utiliser AWS Data Pipeline pour les tâches répétitives de ce type. Pour plus d'informations, consultez [AWS Data Pipeline PigActivity](https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-pigactivity.html)et consultez [AWS Data Pipeline HiveActivity](https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-hiveactivity.html)le *Guide du AWS Data Pipeline développeur*.
# Migration du connecteur Spark Kinesis vers le SDK 2.x pour Amazon EMR 7.0
Le AWS SDK fournit un ensemble complet de bibliothèques permettant d'interagir avec APIs les services de AWS cloud computing, tels que la gestion des informations d'identification, la connexion aux services S3 et Kinesis. Le connecteur Spark Kinesis est utilisé pour consommer les données des flux de données Kinesis, et les données reçues sont transformées et traitées dans le moteur d’exécution de Spark. Actuellement, ce connecteur est basé sur les versions 1.x du AWS SDK et Kinesis-client-library (KCL).
Dans le cadre de la migration vers le AWS SDK 2.x, le connecteur Spark Kinesis est également mis à jour en conséquence pour fonctionner avec le SDK 2.x. Dans la version 7.0 d’Amazon EMR, Spark contient la mise à niveau vers le SDK 2.x, qui n’est pas encore disponible dans la version communautaire d’Apache Spark. Si vous utilisez le connecteur Spark Kinesis sur une version antérieure à la version 7.0, migrez les codes de vos applications pour qu’ils s’exécutent sur le SDK 2.x avant de migrer vers Amazon EMR 7.0.
## Guides de migration
Cette section décrit les étapes de migration d’une application vers la version mise à niveau du connecteur Spark Kinesis. Il inclut des guides de migration vers la bibliothèque client Kinesis (KCL) 2.x, des fournisseurs AWS d'informations d'identification et des clients de AWS service dans le SDK 2.x. AWS À titre de référence, il inclut également un exemple de [WordCount](https://github.com/apache/spark/blob/v3.5.0/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala)programme utilisant le connecteur Kinesis.
**Topics**
+ [Migration de KCL de la version 1.x vers la version 2.x](#migrating-spark-kinesis-KCL-from-1.x-to-2.x)
+ [Migration des fournisseurs AWS d'informations d'identification du AWS SDK 1.x vers le SDK 2.x](#migrating-spark-kinesis-creds-from-1.x-to-2.x)
+ [Migration des clients AWS de service du AWS SDK 1.x vers le SDK 2.x](#migrating-spark-kinesis-service-from-1.x-to-2.x)
+ [Exemples de code pour les applications de streaming](#migrating-spark-kinesis-streaming-examples)
+ [Considérations relatives à l’utilisation de la version mise à niveau du connecteur Spark Kinesis](#migrating-spark-kinesis-considerations)
### Migration de KCL de la version 1.x vers la version 2.x
+ **Niveau et dimensions des métriques dans `KinesisInputDStream`**
Lorsque vous instanciez un `KinesisInputDStream`, vous pouvez contrôler le niveau et les dimensions des métriques du flux. L’exemple suivant montre comment personnaliser ces paramètres avec KCL 1.x :
```
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
val kinesisStream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(kinesisCheckpointInterval)
.storageLevel(StorageLevel.MEMORY_AND_DISK_2)
.metricsLevel(MetricsLevel.DETAILED)
.metricsEnabledDimensions(KinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet)
.build()
```
Dans KCL 2.x, ces paramètres de configuration ont des noms de package différents. Pour migrer vers la version 2.x :
1. Remplacez les instructions d’importation pour `com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration` et `com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel` par `software.amazon.kinesis.metrics.MetricsLevel` et `software.amazon.kinesis.metrics.MetricsUtil` respectivement.
```
// import com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel
import software.amazon.kinesis.metrics.MetricsLevel
// import com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration
import software.amazon.kinesis.metrics.MetricsUtil
```
1. Remplacez la ligne `metricsEnabledDimensionsKinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet` par `metricsEnabledDimensionsSet(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME)`.
Vous trouverez ci-dessous une version mise à jour du `KinesisInputDStream` avec un niveau de métrique et des dimensions de métriques personnalisés :
```
import software.amazon.kinesis.metrics.MetricsLevel
import software.amazon.kinesis.metrics.MetricsUtil
val kinesisStream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(kinesisCheckpointInterval)
.storageLevel(StorageLevel.MEMORY_AND_DISK_2)
.metricsLevel(MetricsLevel.DETAILED)
.metricsEnabledDimensions(Set(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME))
.build()
```
+ Fonction de gestionnaire de messages dans `KinesisInputDStream`
Lors de l’instanciation d’un `KinesisInputDStream`, vous pouvez également spécifier une « fonction de gestionnaire de messages » qui prend un enregistrement Kinesis et renvoie un objet générique T, au cas où vous souhaiteriez utiliser d’autres données incluses dans un enregistrement, comme une clé de partition.
Dans KCL 1.x, la signature de la fonction de gestionnaire de messages est `Record => T`, où Record correspond à `com.amazonaws.services.kinesis.model.Record`. Dans KCL 2.x, la signature du gestionnaire est remplacée par :`KinesisClientRecord => T`, where is. KinesisClientRecord `software.amazon.kinesis.retrieval.KinesisClientRecord`
Voici un exemple de spécification de gestionnaire de messages dans KCL 1.x :
```
import com.amazonaws.services.kinesis.model.Record
def addFive(r: Record): Int = JavaUtils.bytesToString(r.getData).toInt + 5
val stream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.buildWithMessageHandler(addFive)
```
Pour migrer le gestionnaire de messages :
1. Remplacez l’instruction d’importation `com.amazonaws.services.kinesis.model.Record` par `software.amazon.kinesis.retrieval.KinesisClientRecord`.
```
// import com.amazonaws.services.kinesis.model.Record
import software.amazon.kinesis.retrieval.KinesisClientRecord
```
1. Mettez à jour la signature de la méthode du gestionnaire de messages.
```
//def addFive(r: Record): Int = JavaUtils.bytesToString(r.getData).toInt + 5
def addFive = (r: KinesisClientRecord) => JavaUtils.bytesToString(r.data()).toInt + 5
```
Voici un exemple mis à jour de spécification de gestionnaire de messages dans KCL 2.x :
```
import software.amazon.kinesis.retrieval.KinesisClientRecord
def addFive = (r: KinesisClientRecord) => JavaUtils.bytesToString(r.data()).toInt + 5
val stream = KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(Seconds(10))
.storageLevel(StorageLevel.MEMORY_ONLY)
.buildWithMessageHandler(addFive)
```
Pour plus d’informations sur la migration de KCL 1.x vers KCL 2.x, voir la rubrique [Migration des consommateurs de KCL 1.x vers KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html).
### Migration des fournisseurs AWS d'informations d'identification du AWS SDK 1.x vers le SDK 2.x
Les fournisseurs d'informations d'identification sont utilisés pour obtenir des AWS informations d'identification pour les interactions avec AWS. Il existe plusieurs modifications d’interface et de classe liées aux fournisseurs d’informations d’identification dans le SDK 2.x. Ces modifications peuvent être consultées [ici](https://github.com/aws/aws-sdk-java-v2/blob/master/docs/LaunchChangelog.md#122-client-credentials). Le connecteur Spark Kinesis a défini une interface (`org.apache.spark.streaming.kinesis.SparkAWSCredentials`) et des classes d'implémentation qui renvoient la version 1.x des fournisseurs d'informations d' AWS identification. Ces fournisseurs d’informations d’identification sont nécessaires lors de l’initialisation des clients Kinesis. Par exemple, si vous utilisez cette méthode `SparkAWSCredentials.provider` dans les applications, vous devrez mettre à jour les codes pour utiliser la version 2.x des fournisseurs AWS d'informations d'identification.
Voici un exemple d'utilisation des fournisseurs d'informations d'identification dans le AWS SDK 1.x :
```
import org.apache.spark.streaming.kinesis.SparkAWSCredentials
import com.amazonaws.auth.AWSCredentialsProvider
val basicSparkCredentials = SparkAWSCredentials.builder
.basicCredentials("accessKey", "secretKey")
.build()
val credentialProvider = basicSparkCredentials.provider
assert(credentialProvider.isInstanceOf[AWSCredentialsProvider], "Type should be AWSCredentialsProvider")
```
**Pour migrer vers la version 2.x du SDK :**
1. Remplacez l’instruction d’importation `com.amazonaws.auth.AWSCredentialsProvider` par `software.amazon.awssdk.auth.credentials.AwsCredentialsProvider`.
```
//import com.amazonaws.auth.AWSCredentialsProvider
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
```
1. Mettez à jour les autres codes qui utilisent cette classe.
```
import org.apache.spark.streaming.kinesis.SparkAWSCredentials
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
val basicSparkCredentials = SparkAWSCredentials.builder
.basicCredentials("accessKey", "secretKey")
.build()
val credentialProvider = basicSparkCredentials.provider
assert (credentialProvider.isInstanceOf[AwsCredentialsProvider], "Type should be AwsCredentialsProvider")
```
### Migration des clients AWS de service du AWS SDK 1.x vers le SDK 2.x
AWS les clients de service ont des noms de package différents dans la version 2.x (c'est-à-dire`software.amazon.awssdk`), alors que le SDK 1.x les utilise. `com.amazonaws` Pour plus d’informations sur les modifications apportées aux clients, voir [cette page](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-whats-different.html). Si vous utilisez ces clients de service dans les codes, vous devez migrer les clients en conséquence.
Voici un exemple de création d’un client dans la version 1.x du SDK :
```
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
import com.amazonaws.services.dynamodbv2.document.DynamoDB
AmazonDynamoDB ddbClient = AmazonDynamoDBClientBuilder.defaultClient();
AmazonDynamoDBClient ddbClient = new AmazonDynamoDBClient();
```
**Pour migrer vers la version 2.x :**
1. Modifiez les instructions d’importation pour les clients de service. Prenons l’exemple des clients DynamoDB. Remplacez `com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient` ou `com.amazonaws.services.dynamodbv2.document.DynamoDB` par `software.amazon.awssdk.services.dynamodb.DynamoDbClient`.
```
// import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
// import com.amazonaws.services.dynamodbv2.document.DynamoDB
import software.amazon.awssdk.services.dynamodb.DynamoDbClient
```
1. Mettez à jour les codes qui initialisent les clients.
```
// AmazonDynamoDB ddbClient = AmazonDynamoDBClientBuilder.defaultClient();
// AmazonDynamoDBClient ddbClient = new AmazonDynamoDBClient();
DynamoDbClient ddbClient = DynamoDbClient.create();
DynamoDbClient ddbClient = DynamoDbClient.builder().build();
```
Pour plus d'informations sur la migration du AWS SDK de la version 1.x vers la version 2.x, voir [Quelles sont les différences entre le SDK pour AWS Java](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-whats-different.html) 1.x et 2.x ?
### Exemples de code pour les applications de streaming
```
import java.net.URI
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
import software.amazon.awssdk.http.apache.ApacheHttpClient
import software.amazon.awssdk.services.kinesis.KinesisClient
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest
import software.amazon.awssdk.regions.Region
import software.amazon.kinesis.metrics.{MetricsLevel, MetricsUtil}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions
import org.apache.spark.streaming.kinesis.KinesisInitialPositions.Latest
import org.apache.spark.streaming.kinesis.KinesisInputDStream
object KinesisWordCountASLSDKV2 {
def main(args: Array[String]): Unit = {
val appName = "demo-app"
val streamName = "demo-kinesis-test"
val endpointUrl = "https://kinesis.us-west-2.amazonaws.com"
val regionName = "us-west-2"
// Determine the number of shards from the stream using the low-level Kinesis Client
// from the AWS Java SDK.
val credentialsProvider = DefaultCredentialsProvider.create
require(credentialsProvider.resolveCredentials() != null,
"No AWS credentials found. Please specify credentials using one of the methods specified " +
"in https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html")
val kinesisClient = KinesisClient.builder()
.credentialsProvider(credentialsProvider)
.region(Region.US_WEST_2)
.endpointOverride(URI.create(endpointUrl))
.httpClientBuilder(ApacheHttpClient.builder())
.build()
val describeStreamRequest = DescribeStreamRequest.builder()
.streamName(streamName)
.build()
val numShards = kinesisClient.describeStream(describeStreamRequest)
.streamDescription
.shards
.size
// In this example, we are going to create 1 Kinesis Receiver/input DStream for each shard.
// This is not a necessity; if there are less receivers/DStreams than the number of shards,
// then the shards will be automatically distributed among the receivers and each receiver
// will receive data from multiple shards.
val numStreams = numShards
// Spark Streaming batch interval
val batchInterval = Milliseconds(2000)
// Kinesis checkpoint interval is the interval at which the DynamoDB is updated with information
// on sequence number of records that have been received. Same as batchInterval for this
// example.
val kinesisCheckpointInterval = batchInterval
// Setup the SparkConfig and StreamingContext
val sparkConfig = new SparkConf().setAppName("KinesisWordCountASLSDKV2")
val ssc = new StreamingContext(sparkConfig, batchInterval)
// Create the Kinesis DStreams
val kinesisStreams = (0 until numStreams).map { i =>
KinesisInputDStream.builder
.streamingContext(ssc)
.streamName(streamName)
.endpointUrl(endpointUrl)
.regionName(regionName)
.initialPosition(new Latest())
.checkpointAppName(appName)
.checkpointInterval(kinesisCheckpointInterval)
.storageLevel(StorageLevel.MEMORY_AND_DISK_2)
.metricsLevel(MetricsLevel.DETAILED)
.metricsEnabledDimensions(Set(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME))
.build()
}
// Union all the streams
val unionStreams = ssc.union(kinesisStreams)
// Convert each line of Array[Byte] to String, and split into words
val words = unionStreams.flatMap(byteArray => new String(byteArray).split(" "))
// Map each word to a (word, 1) tuple so we can reduce by key to count the words
val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
// Print the first 10 wordCounts
wordCounts.print()
// Start the streaming context and await termination
ssc.start()
ssc.awaitTermination()
}
}
```
### Considérations relatives à l’utilisation de la version mise à niveau du connecteur Spark Kinesis
+ Si vos applications utilisent la `Kinesis-producer-library` avec une version du JDK antérieure à la version 11, des exceptions peuvent se produire, comme `java.lang.NoClassDefFoundError: javax/xml/bind/DatatypeConverter`. Cela est dû au fait que EMR 7.0 est fourni avec le JDK 17 par défaut et que les modules J2EE ont été supprimés des bibliothèques standard depuis Java 11. Il est possible de résoudre le problème en ajoutant la dépendance suivante dans le fichier pom. Remplacez la version de la bibliothèque par une autre selon vos besoins.
```
javax.xml.bind
jaxb-api
${jaxb-api.version}
```
+ Le fichier jar du connecteur Spark Kinesis se trouve à l’emplacement suivant après la création d’un cluster EMR : `/usr/lib/spark/connector/lib/`.
# S3 DistCp (s3-dist-cp)
Apache DistCp est un outil open source que vous pouvez utiliser pour copier de grandes quantités de données. *S3 DistCp* est similaire DistCp, mais optimisé pour fonctionner avec AWS, en particulier, Amazon S3. La commande pour S3 DistCp dans Amazon EMR version 4.0 et versions ultérieures est`s3-dist-cp`, que vous ajoutez en tant qu'étape dans un cluster ou sur la ligne de commande. Avec S3DistCp, vous pouvez copier efficacement de grandes quantités de données d'Amazon S3 vers HDFS où elles peuvent être traitées par les étapes suivantes dans votre cluster Amazon EMR. Vous pouvez également utiliser S3 DistCp pour copier des données entre des compartiments Amazon S3 ou de HDFS vers Amazon S3. S3 DistCp est plus évolutif et plus efficace pour copier en parallèle un grand nombre d'objets entre des compartiments et des AWS comptes.
Pour des commandes spécifiques démontrant la flexibilité de S3DistCP dans des scénarios réels, consultez [Sept conseils d'utilisation de S3 DistCp](https://aws.amazon.com/blogs/big-data/seven-tips-for-using-s3distcp-on-amazon-emr-to-move-data-efficiently-between-hdfs-and-amazon-s3/) sur le AWS blog Big Data.
Par exemple DistCp, S3 DistCp utilise MapReduce pour copier de manière distribuée. Il partage les tâches de copie, de gestion des erreurs, de récupération et de création de rapports entre plusieurs serveurs. Pour plus d'informations sur le projet DistCp open source Apache, consultez le [DistCpguide](http://hadoop.apache.org/docs/stable/hadoop-distcp/DistCp.html) dans la documentation d'Apache Hadoop.
Si S3 DistCp ne parvient pas à copier tout ou partie des fichiers spécifiés, l'étape du cluster échoue et renvoie un code d'erreur différent de zéro. Dans ce cas, S3 DistCp ne nettoie pas les fichiers partiellement copiés.
**Important**
S3 DistCp ne prend pas en charge les noms de compartiment Amazon S3 contenant le caractère de soulignement.
S3 DistCp ne prend pas en charge la concaténation pour les fichiers Parquet. Utilisez PySpark plutôt. Pour plus d'informations, consultez [Concaténation de fichiers parquet dans Amazon EMR](https://aws.amazon.com/premiumsupport/knowledge-center/emr-concatenate-parquet-files/).
Pour éviter les erreurs de copie lorsque vous utilisez S3DistCp pour copier un seul fichier (au lieu d'un répertoire) de S3 vers HDFS, utilisez Amazon EMR version 5.33.0 ou ultérieure, ou Amazon EMR version 6.3.0 ou ultérieure.
## DistCp Options S3
Bien que similaire DistCp, S3 DistCp prend en charge un ensemble d'options différent pour modifier la façon dont il copie et compresse les données.
Lorsque vous appelez S3DistCp, vous pouvez spécifier les options décrites dans le tableau suivant. Les options sont ajoutées à l'étape à l'aide de la liste d'arguments. Des exemples d'DistCp arguments S3 sont présentés dans le tableau suivant.
| Option | Description | Obligatoire |
| --- | --- | --- |
| ‑‑src=LOCATION | Emplacement des données à copier. Cela peut être un emplacement dans HDFS ou Amazon S3. Exemple : `‑‑src=s3://amzn-s3-demo-bucket/logs/j-3GYXXXXXX9IOJ/node` S3 DistCp ne prend pas en charge les noms de compartiment Amazon S3 contenant le caractère de soulignement. | Oui |
| ‑‑dest=LOCATION | Destination des données. Cela peut être un emplacement dans HDFS ou Amazon S3. Exemple : `‑‑dest=hdfs:///output` S3 DistCp ne prend pas en charge les noms de compartiment Amazon S3 contenant le caractère de soulignement. | Oui |
| ‑‑srcPattern=PATTERN | [Expression régulière](http://en.wikipedia.org/wiki/Regular_expression) qui filtre l'opération de copie sur un sous-ensemble des données à l'adresse `‑‑src`. Si `‑‑srcPattern` et `‑‑groupBy` ne sont pas spécifiés, toutes les données à l'adresse `‑‑src` sont copiées dans `‑‑dest`. Si l'argument de l'expression régulière contient des caractères spéciaux, tels qu'un astérisque (\$1), l'expression régulière ou la chaîne `‑‑args` complète doit être mise entre guillemets simples ('). Exemple : `‑‑srcPattern=.*daemons.*-hadoop-.*` | Non |
| ‑‑groupBy=PATTERN | [Expression régulière](http://en.wikipedia.org/wiki/Regular_expression) qui oblige S3 DistCp à concaténer les fichiers correspondant à l'expression. Par exemple, vous pouvez utiliser cette option pour combiner tous les fichiers journaux écrits en une heure dans un fichier unique. Le nom de fichier concaténé est la valeur correspondante à l'expression régulière pour le regroupement. Les parenthèses indiquent comment les fichiers doivent être regroupés, en combinant tous les éléments qui correspondent à l'instruction entre parenthèses en un fichier de sortie unique. Si l'expression régulière n'inclut pas d'instruction entre parenthèses, le cluster échoue à l'DistCp étape S3 et renvoie une erreur. Si l'argument de l'expression régulière contient des caractères spéciaux, tels qu'un astérisque (\$1), l'expression régulière ou la chaîne `‑‑args` complète doit être mise entre guillemets simples ('). Lorsque `‑‑groupBy` est spécifié, seuls les fichiers qui correspondent au modèle spécifié sont copiés. Vous n'avez pas besoin de spécifier `‑‑groupBy` et `‑‑srcPattern` en même temps. Exemple : `‑‑groupBy=.*subnetid.*([0-9]+-[0-9]+-[0-9]+-[0-9]+).*` | Non |
| ‑‑targetSize=SIZE | Taille, en mébioctets (Mio), des fichiers à créer en fonction de l'option `‑‑groupBy`. Cette valeur doit être un nombre entier. Lorsque cette option `‑‑targetSize` est définie, S3 DistCp essaie de correspondre à cette taille ; la taille réelle des fichiers copiés peut être supérieure ou inférieure à cette valeur. Les travaux sont regroupés en fonction de la taille du fichier de données. Par conséquent, il est possible que la taille du fichier cible corresponde à la taille du fichier de données source. Si les fichiers concaténés par `‑‑groupBy` sont plus grands que la valeur de `‑‑targetSize`, ils sont divisés en fichiers plus petits, nommés de manière séquentielle avec une valeur numérique ajoutée à la fin de leur nom. Par exemple, un fichier concaténé `myfile.gz` peut être divisé en parties telles que : `myfile0.gz`, `myfile1.gz`, etc. Exemple : `‑‑targetSize=2` | Non |
| ‑‑appendToLastFile | Spécifie le comportement de S3 DistCp lors de la copie vers des fichiers déjà présents depuis Amazon S3 vers HDFS. Il ajoute les données des nouveaux fichiers dans les fichiers existants. Si vous utilisez `‑‑appendToLastFile` avec `‑‑groupBy`, les nouvelles données sont ajoutées aux fichiers qui correspondent aux mêmes groupes. Cette option respecte également le comportement `‑‑targetSize` lorsqu'il est utilisé avec `‑‑groupBy.` | Non |
| ‑‑outputCodec=CODEC | Spécifie le codec de compression à utiliser pour les fichiers copiés. Cela peut prendre les valeurs : `gzip`, `gz`, `lzo`, `snappy` ou `none`. Vous pouvez utiliser cette option, par exemple, pour convertir des fichiers d'entrée compressés avec Gzip en fichiers de sortie avec la compression LZO, ou pour décompresser les fichiers dans le cadre de l'opération de copie. Si vous choisissez un codec de sortie, le nom de fichier sera ajouté avec l'extension appropriée (par exemple, pour `gz` et `gzip`, l'extension est `.gz`). Si vous ne spécifiez pas de valeur pour `‑‑outputCodec`, les fichiers sont copiés sans modification dans leur compression. Exemple : `‑‑outputCodec=lzo` | Non |
| ‑‑s3ServerSideEncryption | Garantit que les données cibles sont transférées via SSL et automatiquement chiffrées dans Amazon S3 à l'aide d'une AWS clé côté service. Lorsque vous récupérez des données à l'aide de S3DistCp, les objets sont automatiquement déchiffrés. Si vous tentez de copier un objet non chiffré vers un compartiment Amazon S3 requérant le chiffrement, l'opération échoue. Pour plus d'informations, consultez [Utilisation du chiffrement des données](https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingEncryption.html). Exemple : `‑‑s3ServerSideEncryption` | Non |
| ‑‑deleteOnSuccess | Si l'opération de copie est réussie, cette option oblige S3 DistCp à supprimer les fichiers copiés de l'emplacement source. Cela est utile si vous copiez des fichiers de sortie, tels que les fichiers journaux, d'un emplacement à un autre comme une tâche planifiée, et que vous ne voulez pas copier deux fois les mêmes fichiers. Exemple : `‑‑deleteOnSuccess` | Non |
| ‑‑disableMultipartUpload | Désactive l'utilisation du téléchargement partitionné. Exemple : `‑‑disableMultipartUpload` | Non |
| ‑‑multipartUploadChunkSize=SIZE | Taille, en Mio, de chaque partie d'un chargement partitionné sur Amazon S3. S3 DistCp utilise le téléchargement partitionné lorsqu'il copie des données dont la `multipartUploadChunkSize` taille est supérieure à. Pour améliorer la performance au travail, vous pouvez augmenter la taille de chaque pièce. La taille par défaut est de 128 Mio. Exemple : `‑‑multipartUploadChunkSize=1000` | Non |
| ‑‑numberFiles | Ajoute aux fichiers de sortie des numéros séquentiels. Le compteur démarre à 0, à moins qu'une autre valeur soit spécifiée par `‑‑startingIndex`. Exemple : `‑‑numberFiles` | Non |
| ‑‑startingIndex=INDEX | Utilisé avec `‑‑numberFiles` pour spécifier le premier numéro de la séquence. Exemple : `‑‑startingIndex=1` | Non |
| ‑‑outputManifest=FILENAME | Crée un fichier texte, compressé avec Gzip, qui contient une liste de tous les fichiers copiés par S3DistCp. Exemple : `‑‑outputManifest=manifest-1.gz` | Non |
| ‑‑previousManifest=PATH | Lit un fichier manifeste créé lors d'un appel précédent à S3 à DistCp l'aide de l'`‑‑outputManifest`indicateur. Lorsque l'`‑‑previousManifest`indicateur est défini, S3 DistCp exclut les fichiers répertoriés dans le manifeste de l'opération de copie. Si `‑‑outputManifest` est spécifié avec `‑‑previousManifest`, les fichiers répertoriés dans le manifeste précédent apparaissent également dans le nouveau fichier manifeste, même si les fichiers ne sont pas copiés. Exemple : `‑‑previousManifest=/usr/bin/manifest-1.gz` | Non |
| ‑‑requirePreviousManifest | Nécessite un manifeste créé lors d'un appel précédent à S3DistCp. Si la valeur spécifiée est false, aucune erreur n'est générée lorsqu'un manifeste précédent n'est pas spécifié. Par défaut, la valeur est true. | Non |
| ‑‑copyFromManifest | Inverse le comportement de S3 `‑‑previousManifest` DistCp pour qu'il utilise le fichier manifeste spécifié comme liste de fichiers à copier, au lieu d'une liste de fichiers à exclure de la copie. Exemple : `‑‑copyFromManifest ‑‑previousManifest=/usr/bin/manifest-1.gz` | Non |
| ‑‑s3Endpoint=ENDPOINT | Spécifie le point de terminaison Amazon S3 à utiliser lors du chargement d'un fichier. Cette option définit le point de terminaison pour la source et la destination. S'il n'est pas défini, le point de terminaison par défaut est `s3.amazonaws.com`. Pour obtenir la liste des points de terminaison Amazon S3, consultez [Régions et points de terminaison](https://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region). Exemple : `‑‑s3Endpoint=s3.eu-west-1.amazonaws.com` | Non |
| ‑‑storageClass=CLASS | La classe de stockage à utiliser lorsque la destination est Amazon S3. Les valeurs valides sont STANDARD et REDUCED\$1REDUNDANCY. Si cette option n'est pas spécifiée, S3 DistCp essaie de préserver la classe de stockage. Exemple : `‑‑storageClass=STANDARD` | Non |
| ‑‑srcPrefixesFile=PATH | fichier texte dans Amazon S3 (s3://), HDFS (hdfs:///) ou le système de fichiers local (file:/) qui contient une liste de préfixes `src`, un préfixe par ligne. S'il `srcPrefixesFile` est fourni, S3 ne DistCp listera pas le chemin src. Au lieu de cela, il génère une liste source en tant que résultat combiné de la liste de tous les préfixes spécifiés dans ce fichier. Le chemin d'accès relatif par rapport au chemin d'accès src, à la place de ces préfixes, sera utilisé pour générer les chemins de destination. Si `srcPattern` est également spécifié, il est appliqué aux résultats combinés de liste des préfixes sources pour filtrer encore plus les données d'entrée. Si `copyFromManifest` est utilisé, les objets dans le manifeste seront copiés et `srcPrefixesFile` sera ignoré. Exemple : `‑‑srcPrefixesFile=PATH` | Non |
Outre les options ci-dessus, S3 DistCp implémente l'[interface Tool](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/util/Tool.html), ce qui signifie qu'elle prend en charge les options génériques.
## Ajouter S3 DistCp en tant qu'étape dans un cluster
Vous pouvez appeler S3 DistCp en l'ajoutant en tant qu'étape dans votre cluster. Il est possible d'ajouter des étapes à un cluster lors de son lancement ou à un cluster en cours d'exécution à l'aide de la console, de l'interface de ligne de commande ou de l'API. Les exemples suivants illustrent l'ajout d'une DistCp étape S3 à un cluster en cours d'exécution. Pour plus d'informations sur l'ajout d'étapes dans un cluster, consultez [Soumission de travail à un cluster](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-work-with-steps.html) dans le *Guide de gestion d'Amazon EMR*.
**Pour ajouter une DistCp étape S3 à un cluster en cours d'exécution à l'aide du AWS CLI**
Pour plus d'informations sur l'utilisation des commandes Amazon EMR dans le AWS CLI, consultez la référence des [AWS CLI commandes](https://docs.aws.amazon.com/cli/latest/reference/emr).
+ Pour ajouter une étape à un cluster qui appelle S3DistCp, transmettez les paramètres qui spécifient comment S3 DistCp doit effectuer l'opération de copie en tant qu'arguments.
L'exemple suivant copie les journaux des démons de Amazon S3 dans `hdfs:///output`. Dans la commande suivante :
+ `‑‑cluster-id` spécifie le cluster
+ `Jar`est l'emplacement du fichier DistCp JAR S3. Pour un exemple d'exécution d'une commande sur un cluster à l'aide de command-runner.jar, consultez [Soumettre une étape JAR personnalisée pour exécuter un script ou une commande](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-commandrunner.html#emr-commandrunner-examples).
+ `Args`est une liste séparée par des virgules des paires nom-valeur des options à transmettre à S3. DistCp Pour obtenir la liste complète des options disponibles, consultez [DistCp Options S3](#UsingEMR_s3distcp.options).
Pour ajouter une étape de DistCp copie S3 à un cluster en cours d'exécution, placez ce qui suit dans un fichier JSON enregistré dans Amazon S3 ou dans votre système de fichiers local, comme dans cet `myStep.json` exemple. *j-3GYXXXXXX9IOK*Remplacez-le par votre ID de cluster et remplacez *amzn-s3-demo-bucket* par le nom de votre compartiment Amazon S3.
```
[
{
"Name":"S3DistCp step",
"Args":["s3-dist-cp","‑‑s3Endpoint=s3.amazonaws.com","‑‑src=s3://amzn-s3-demo-bucket/logs/j-3GYXXXXXX9IOJ/node/","‑‑dest=hdfs:///output","‑‑srcPattern=.*[a-zA-Z,]+"],
"ActionOnFailure":"CONTINUE",
"Type":"CUSTOM_JAR",
"Jar":"command-runner.jar"
}
]
```
```
aws emr add-steps ‑‑cluster-id j-3GYXXXXXX9IOK ‑‑steps file://./myStep.json
```
**Example Copie des fichiers journaux d'Amazon S3 vers HDFS**
Cet exemple illustre également comment copier les fichiers journaux stockés dans un compartiment Amazon S3 vers HDFS en ajoutant une étape à un cluster en cours d'exécution. Dans cet exemple, l'option `‑‑srcPattern` est utilisée pour limiter les données copiées dans les journaux des démons.
Pour copier les fichiers journaux d'Amazon S3 vers HDFS à l'aide de l'option `‑‑srcPattern`, placez ce qui suit dans un fichier JSON enregistré dans Amazon S3 ou dans votre système de fichiers local en tant que `myStep.json` pour cet exemple. *j-3GYXXXXXX9IOK*Remplacez-le par votre ID de cluster et remplacez *amzn-s3-demo-bucket* par le nom de votre compartiment Amazon S3.
```
[
{
"Name":"S3DistCp step",
"Args":["s3-dist-cp","‑‑s3Endpoint=s3.amazonaws.com","‑‑src=s3://amzn-s3-demo-bucket/logs/j-3GYXXXXXX9IOJ/node/","‑‑dest=hdfs:///output","‑‑srcPattern=.*daemons.*-hadoop-.*"],
"ActionOnFailure":"CONTINUE",
"Type":"CUSTOM_JAR",
"Jar":"command-runner.jar"
}
]
```
## Nettoyage après l'échec des DistCp tâches S3
Si S3 DistCp ne parvient pas à copier certains ou tous les fichiers spécifiés, l'étape de commande ou de cluster échoue et renvoie un code d'erreur différent de zéro. Dans ce cas, S3 DistCp ne nettoie pas les fichiers partiellement copiés. Vous devez les supprimer manuellement.
Les fichiers partiellement copiés sont enregistrés dans le `tmp` répertoire HDFS dans des sous-répertoires avec l'identifiant unique de la tâche S3DistCp . Cet identifiant figure dans la sortie standard de la tâche.
Par exemple, pour une DistCp tâche S3 avec l'ID`4b1c37bb-91af-4391-aaf8-46a6067085a6`, vous pouvez vous connecter au nœud principal du cluster et exécuter la commande suivante pour afficher les fichiers de sortie associés à la tâche.
```
hdfs dfs -ls /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output
```
La commande renvoie une liste de fichiers similaires à la liste suivante :
```
Found 8 items
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/_SUCCESS
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:02 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00000
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:02 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00001
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:02 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00002
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00003
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00004
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00005
-rw-r‑‑r‑‑ 1 hadoop hadoop 0 2018-12-10 06:03 /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output/part-r-00006
```
Vous pouvez ensuite exécuter la commande suivante pour supprimer le répertoire et l'ensemble du contenu.
```
hdfs dfs rm -rf /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6
```