Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
# Conectores y utilidades
Amazon EMR proporciona varios conectores y utilidades para acceder a otros AWS servicios como fuentes de datos. Normalmente, es posible acceder a los datos de estos servicios dentro de un programa. Por ejemplo, puede especificar una transmisión de Kinesis en una consulta de Hive, un script de Pig o una MapReduce aplicación y, a continuación, operar con esos datos.
**Topics**
+ [Exportación, importación, consulta y unión de tablas en DynamoDB mediante Amazon EMR](EMRforDynamoDB.md)
+ [Kinesis](emr-kinesis.md)
+ [S3 (s3-dist-cp) DistCp](UsingEMR_s3distcp.md)
+ [Limpiando después de DistCp trabajos fallidos en S3](#s3distcp-cleanup)
# Exportación, importación, consulta y unión de tablas en DynamoDB mediante Amazon EMR
**nota**
El conector Amazon EMR-DynamoDB es de código abierto en. GitHub Para obtener más información, consulte [https://github.com/awslabs/emr-dynamodb-connector](https://github.com/awslabs/emr-dynamodb-connector).
DynamoDB es un servicio de bases de datos NoSQL totalmente administrado que proporciona un rendimiento rápido y predecible, así como una perfecta escalabilidad. Los desarrolladores pueden crear una tabla de base de datos y ampliar su tráfico de solicitudes o almacenamiento sin límite. DynamoDB distribuye automáticamente los datos y el tráfico para la tabla entre un número de servidores suficiente para administrar la capacidad de solicitudes que el cliente especifique y la cantidad de datos almacenados, manteniendo al mismo tiempo un rendimiento uniforme y rápido. Mediante Amazon EMR y Hive puede procesar de forma rápida y eficaz grandes cantidades de datos, como, por ejemplo, datos almacenados en DynamoDB. Para obtener más información sobre DynamoDB, consulte la [Guía para desarrolladores de Amazon DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/).
Apache Hive es una capa de software que puede utilizar para consultar clústeres de MapReduce utilizando un lenguaje de consultas de tipo SQL simplificado denominado HiveQL. Se ejecuta sobre la arquitectura de Hadoop. Para obtener más información acerca de Hive y HiveQL, consulte el [HiveQL Language Manual](https://cwiki.apache.org/confluence/display/Hive/LanguageManual). Para obtener más información sobre Hive y Amazon EMR, consulte [Apache Hive](emr-hive.md).
Puede utilizar Amazon EMR con una versión personalizada de Hive que incluye conectividad con DynamoDB para realizar operaciones con los datos almacenados en DynamoDB:
+ Cargar datos de DynamoDB en Sistema de archivos distribuido de Hadoop (HDFS) y utilizarlos como entrada en un clúster de Amazon EMR.
+ Consultar datos de DynamoDB en directo utilizando instrucciones similares a SQL (HiveQL).
+ Unir datos almacenados en DynamoDB y exportarlos o consultar respecto a los datos unidos.
+ Exportación de datos almacenados en DynamoDB a Amazon S3.
+ Importación de datos almacenados en Amazon S3 a DynamoDB.
**nota**
- El conector Amazon EMR-DynamoDB no es compatible con clústeres configurados para utilizar la [autenticación Kerberos](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-kerberos.html).
Para realizar cada una de las siguientes tareas, lanzará un clúster de Amazon EMR, especificará la ubicación de los datos en DynamoDB y emitirá comandos de Hive para manipular los datos en DynamoDB.
Hay varias formas de lanzar un clúster de Amazon EMR: puede utilizar la consola de Amazon EMR, la interfaz de línea de comandos (CLI) o puede programar el clúster mediante un SDK AWS o la API de Amazon EMR. También puede elegir si desea ejecutar un clúster de Hive de forma interactiva o a partir de un script. En esta sección le mostraremos cómo lanzar un clúster de Hive interactivo desde la consola de Amazon EMR y en la CLI.
Utilizar Hive de forma interactiva es una forma fantástica de probar el rendimiento de las consultas y ajustar su aplicación. Después de que haya establecido un conjunto de comandos de Hive que se ejecutarán de forma periódica, considere la posibilidad de crear un script de Hive que Amazon EMR puede ejecutar por usted.
**aviso**
Las operaciones de lectura o escritura de Amazon EMR en una tabla de DynamoDB se tienen en cuenta para el rendimiento aprovisionado establecido, por lo que es posible que aumente la frecuencia de las excepciones de rendimiento aprovisionado. En el caso de solicitudes de gran tamaño, Amazon EMR implementa los reintentos con retroceso exponencial para administrar la carga de solicitudes en la tabla de DynamoDB. La ejecución de tareas de Amazon EMR de forma simultánea con otro tráfico podría provocar que supere el nivel de rendimiento aprovisionado asignado. Puedes monitorizarlo comprobando la **ThrottleRequests**métrica en Amazon CloudWatch. Si la carga de solicitudes es demasiado elevada, puede volver a lanzar el clúster y definir [Ajuste de porcentaje de lectura](EMR_Hive_Optimizing.md#ReadPercent) o [Configuración de porcentaje de escritura](EMR_Hive_Optimizing.md#WritePercent) en un valor inferior para limitar las operaciones de Amazon EMR. Para obtener información sobre la configuración de rendimiento de DynamoDB, consulte [Rendimiento aprovisionado](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithDDTables.html#ProvisionedThroughput).
Si una tabla está configurada para el [modo bajo demanda](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.ReadWriteCapacityMode.html#HowItWorks.OnDemand), debe volver a cambiarlo al modo aprovisionado antes de ejecutar una operación de exportación o importación. Las tuberías necesitan una relación de rendimiento para calcular los recursos que se van a utilizar de un Dynamo. DBtable El modo bajo demanda elimina el rendimiento aprovisionado. Para aprovisionar la capacidad de rendimiento, puede utilizar las métricas de Amazon CloudWatch Events para evaluar el rendimiento agregado que ha utilizado una tabla.
**Topics**
+ [Configurar una tabla de Hive para ejecutar comandos de Hive](EMR_Interactive_Hive.md)
+ [Ejemplo de comandos de Hive para la exportación, importación y consulta de datos en DynamoDB](EMR_Hive_Commands.md)
+ [Optimización de rendimiento para las operaciones de Amazon EMR en DynamoDB](EMR_Hive_Optimizing.md)
# Configurar una tabla de Hive para ejecutar comandos de Hive
Apache Hive es una aplicación de almacenamiento de datos que se puede utilizar para consultar los datos contenidos en es de Amazon EMR con un lenguaje de tipo SQL. Para obtener más información acerca de Hive, consulte [http://hive.apache.org](http://hive.apache.org/).
El siguiente procedimiento supone que ya ha creado un clúster y especificado un par de claves de Amazon EC2. Para obtener información sobre cómo empezar a crear clústeres, consulte [Introducción a Amazon EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-gs) en la *Guía de administración de Amazon EMR*.
## Configure Hive para usar MapReduce
Cuando se usa Hive en Amazon EMR para consultar tablas de DynamoDB, se pueden producir errores si Hive usa el motor de ejecución predeterminado, Tez. Por este motivo, cuando cree un clúster con Hive que se integre con DynamoDB, tal como se describe en esta sección, le recomendamos que utilice una clasificación de configuración que defina el uso de Hive. MapReduce Para obtener más información, consulte [Configuración de aplicaciones](emr-configure-apps.md).
En el siguiente fragmento se muestran la clasificación y la propiedad de la configuración que se van a utilizar para configurarla MapReduce como motor de ejecución de Hive:
```
[
{
"Classification": "hive-site",
"Properties": {
"hive.execution.engine": "mr"
}
}
]
```
**Para ejecutar comandos de Hive de forma interactiva**
1. Conecte al nodo principal. Para obtener más información, consulte [Conexión al nodo maestro mediante SSH](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html) en la *Guía de administración de Amazon EMR*.
1. En el símbolo del sistema de comandos para el nodo principal, escriba `hive`.
Debería ver un símbolo del sistema de Hive: `hive>`
1. Escriba un comando de Hive que asigna una tabla en la aplicación de Hive a los datos de DynamoDB. Esta tabla actúa como una referencia para los datos almacenados en Amazon DynamoDB; los datos no se almacenan localmente en Hive y las consultas que utilizan esta tabla se ejecutan frente a los datos en vivo en DynamoDB, consumiendo capacidad de lectura o escritura de la tabla cada vez que se ejecuta un comando. Si espera ejecutar varios comandos de Hive respecto al mismo conjunto de datos, plantéese en primer lugar la opción de exportar.
A continuación, se muestra la sintaxis para el mapeo de una tabla de Hive a una tabla de 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...");
```
Al crear una tabla en Hive desde DynamoDB, debe crearla como tabla externa mediante la palabra clave `EXTERNAL`. La diferencia entre las tablas internas y externas es que los datos en tablas internas se eliminan cuando se anula una tabla interna. Este no es el comportamiento deseado cuando hay conexión a Amazon DynamoDB y, por lo tanto, solo se admiten las tablas externas.
Por ejemplo, el siguiente comando de Hive crea una tabla denominada *hivetable1* en Hive que hace referencia a la tabla de DynamoDB denominada *dynamodbtable1*. La *tabla* dynamodbtable1 de DynamoDB tiene un esquema de clave principal. hash-and-range El elemento de clave hash es `name` (tipo cadena), el elemento de clave de rango es `year` (tipo numérico) y cada elemento tiene un valor de atributo para `holidays` (tipo conjunto de cadenas).
```
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 línea 1 utiliza la instrucción `CREATE EXTERNAL TABLE` de HiveQL. Para *hivetable1*, debe establecer una columna para cada par de atributos nombre-valor en la tabla de DynamoDB y proporcionar el tipo de datos. Estos valores no distinguen entre mayúsculas y minúsculas y puede proporcionar a las columnas cualquier nombre (excepto palabras reservadas).
La línea 2 utiliza la instrucción `STORED BY`. El valor `STORED BY` es el nombre de la clase que se encarga de la conexión entre Hive y DynamoDB. Debe definirse en `'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'`.
La línea 3 utiliza la instrucción `TBLPROPERTIES` para asociar "hivetable1" con la tabla y el esquema correctos en DynamoDB. Proporcione a `TBLPROPERTIES` valores para el parámetro `dynamodb.table.name` y el parámetro `dynamodb.column.mapping`. Estos valores *distinguen* entre mayúsculas y minúsculas.
**nota**
Todos los nombres de atributos de DynamoDB de la tabla deben tener las columnas correspondientes en la tabla de Hive. Según la versión de Amazon EMR, se producen las siguientes situaciones si el one-to-one mapeo no existe:
En la versión 5.27.0 y posteriores de Amazon EMR, el conector tiene validaciones que garantizan un mapeo one-to-one entre los nombres de los atributos de DynamoDB y las columnas de la tabla Hive. Si el mapeo no existe, se producirá un error. one-to-one
En la versión 5.26.0 de Amazon EMR y anteriores, la tabla de Hive no incluirá el par nombre-valor de DynamoDB. Si no asigna los atributos de clave principal de DynamoDB, Hive devuelve un error. Si no asigna un atributo de clave no principal, no se genera ningún error, pero no verá los datos en la tabla de Hive. Si los tipos de datos no coinciden, el valor es nulo.
A continuación, puede comenzar a ejecutar operaciones de Hive en *hivetable1*. Las consultas que se ejecutan en *hivetable1* se ejecutan internamente en la tabla de DynamoDB *dynamodbtable1* de su cuenta de DynamoDB, y consumen unidades de lectura o escritura con cada ejecución.
Cuando ejecuta consultas de Hive frente a una tabla de DynamoDB, debe asegurarse de que ha aprovisionado una cantidad suficiente de unidades de capacidad de lectura.
Por ejemplo, suponga que ha aprovisionado 100 unidades de capacidad de lectura para su tabla de DynamoDB. Esto le permite realizar 100 lecturas o 409 600 bytes, por segundo. Si dicha tabla contiene 20 GB de datos (21 474 836 480 bytes) y la consulta de Hive realiza un análisis completo de la tabla, puede calcular aproximadamente cuánto tardará la consulta en ejecutarse:
* 21 474 836 480 / 409 600 = 52 429 segundos = 14,56 horas *
La única forma de reducir el tiempo necesario, sería ajustar las unidades de capacidad de lectura en el origen de la tabla de DynamoDB. Agregar más nodos de Amazon EMR no le ayudará.
En la salida de Hive, el porcentaje completado se actualiza cuando finalizan uno o varios procesos de mapeador. Para una tabla de DynamoDB grande con un ajuste bajo de capacidad de lectura aprovisionada, la salida de porcentaje de finalización podría no actualizarse durante mucho tiempo; en el caso anterior, el trabajo aparecerá con un 0% completo durante varias horas. Para ver un estado más detallado del progreso del trabajo, acceda a la consola de Amazon EMR; podrá ver el estado de la tarea de asignador individual y estadísticas de lecturas de datos. También puede iniciar sesión en la interfaz de usuario de Hadoop en el nodo principal y ver las estadísticas de Hadoop. Esto le mostrará el estado de tareas de mapeo individuales y algunos datos de estadísticas de lectura. Para obtener más información, consulte los temas siguientes:
+ [Interfaces web alojadas en el nodo maestro](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html)
+ [Ver las interfaces web de Hadoop](https://docs.aws.amazon.com/emr/latest/ManagementGuide/UsingtheHadoopUserInterface.html)
Para obtener más información sobre las instrucciones de HiveQL de ejemplo para realizar tareas tales como la exportación o importación de datos desde DynamoDB y la unión de tablas, consulte [Ejemplo de comandos de Hive para la exportación, importación y consulta de datos en DynamoDB](EMR_Hive_Commands.md).
**Para cancelar una solicitud de Hive**
Cuando se ejecuta una consulta de Hive, la respuesta inicial del servidor incluye el comando para cancelar la solicitud. Para cancelar la solicitud en cualquier momento del proceso, utilice **Kill Command (Comando de cancelación)** de la respuesta del servidor.
1. Introduzca `Ctrl+C` para salir del cliente de línea de comando.
1. En el símbolo del shell, introduzca el **Kill Command (Comando de cancelación)** de la respuesta inicial del servidor a su solicitud.
Como alternativa, puede ejecutar el siguiente comando desde la línea de comandos del nodo maestro para eliminar el trabajo de Hadoop, que *job-id* es el identificador del trabajo de Hadoop y se puede recuperar desde la interfaz de usuario de Hadoop.
```
hadoop job -kill job-id
```
## Tipos de datos para Hive y DynamoDB
En la siguiente tabla se muestran los tipos de datos de Hive disponibles, el tipo predeterminado de DynamoDB al que se corresponden y los tipos alternativos de DynamoDB a los que también se pueden asignar.
| Tipo de Hive | Tipo predeterminado de DynamoDB | Tipos alternativos de DynamoDB |
| --- | --- | --- |
| cadena | cadena (S) | |
| bigint o doble | número (N) | |
| binario | binario (B) | |
| booleano | booleano (BOOL) | |
| array | lista (L) | conjunto de números (NS), un conjunto de cadenas (SS) o un conjunto de binarios (BS) |
| mapa | elemento | mapa (M) |
| mapa | mapa (M) | |
| | nulo (NULL) | |
Si desea escribir los datos de Hive como un correspondiente tipo alternativo de DynamoDB o si los datos de DynamoDB incluyen valores de atributo de un tipo alternativo de DynamoDB, puede especificar la columna y el tipo de DynamoDB con el parámetro `dynamodb.type.mapping`. En el siguiente ejemplo se muestra la sintaxis para especificar un mapeo de tipo alternativo.
```
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");
```
El parámetro de mapeo de tipos es opcional y solo tiene que especificarse para las columnas que utilizan tipos alternativos.
Por ejemplo, el siguiente comando de Hive crea una tabla denominada `hivetable2` que hace referencia a la tabla `dynamodbtable2` de DynamoDB. Es similar a `hivetable1`, salvo que asigna la columna `col3` al tipo de conjunto de cadenas (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");
```
En Hive, `hivetable1` y `hivetable2` son idénticos. Sin embargo, cuando los datos de esas tablas se escriben en sus correspondientes tablas de DynamoDB, `dynamodbtable1` incluirá listas, mientras que `dynamodbtable2` incluirá conjuntos de cadenas.
Si desea escribir valores `null` de Hive como atributos de tipo `null` de DynamoDB, puede hacerlo con el parámetro `dynamodb.null.serialization`. En el siguiente ejemplo se muestra la sintaxis para especificar la serialización `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");
```
El parámetro de serialización null es opcional y se establece en `false` si no se especifica. Tenga en cuenta que los atributos `null` de DynamoDB se leen como valores `null` en Hive con independencia de la configuración del parámetro. Las colecciones de Hive con valores `null` solo pueden escribirse en DynamoDB si el parámetro de serialización null se especifica como `true`. De lo contrario, se produce un error de Hive.
El tipo bigint en Hive es igual al tipo long en Java y el tipo de doble de Hive es igual que el tipo doble de Java en términos de precisión. Esto significa que, si dispone de datos numéricos almacenados en DynamoDB con precisión superior a la disponible en los tipos de datos de Hive, el uso de Hive para exportar, importar o hacer referencia a los datos de DynamoDB podría dar lugar a una pérdida de precisión o un error en la consulta de Hive.
Las exportaciones del tipo binario desde DynamoDB a Amazon Simple Storage Service (Amazon S3) o HDFS se almacenan como cadena codificada en Base64. Si está importando datos desde Amazon S3 o HDFS al tipo binario de DynamoDB, deberían estar codificados como una cadena Base64.
## Opciones de Hive
Puede definir las siguientes opciones de Hive para gestionar la transferencia de datos fuera de Amazon DynamoDB. Estas opciones solo persisten para la sesión de Hive actual. Si cierra el símbolo del sistema de Hive y vuelve a abrirlo más adelante en el clúster, estos ajustes habrán vuelto a los valores predeterminados.
| Opciones de Hive | Description (Descripción) |
| --- | --- |
| dynamodb.throughput.read.percent | Define la velocidad de las operaciones de lectura para mantener la tasa de rendimiento aprovisionada de DynamoDB en el rango asignado para la tabla. El valor está entre `0.1` y `1.5`, ambos incluidos. El valor de 0,5 es la tasa de lectura predeterminada, lo que significa que Hive intentará consumir la mitad de la lectura aprovisionada a través de los recursos de la tabla. Al aumentar este valor por encima de 0,5 se aumenta la tasa de solicitud de lectura. Si se reduce el valor por debajo de 0,5 se reduce la tasa de solicitud de lectura. Esta tasa de lectura es aproximada. La tasa de lectura real dependerá de factores tales como el hecho de que exista una distribución uniforme de claves en DynamoDB. Si encuentra que la operación de Hive suele superar el rendimiento aprovisionado o si el tráfico de lectura en directo se está demasiado limitado, entonces reduzca este valor por debajo de `0.5`. Si dispone de suficiente capacidad y desea una operación de Hive más rápida, defina este valor por encima de `0.5`. También puede suscribir en exceso definiendo este valor en 1,5 si cree que hay operaciones de entrada/salida sin utilizar disponibles. |
| dynamodb.throughput.write.percent | Establece la velocidad de las operaciones de escritura para mantener la tasa de rendimiento aprovisionada de DynamoDB en el rango asignado para la tabla. El valor está entre `0.1` y `1.5`, ambos incluidos. El valor de 0,5 es la tasa de escritura predeterminada, lo que significa que Hive intentará consumir la mitad de la escritura aprovisionada a través de los recursos de la tabla. Al aumentar este valor por encima de 0,5 se aumenta la tasa de solicitud de escritura. Si se reduce el valor por debajo de 0,5 se reduce la tasa de solicitud de escritura. Esta tasa de escritura es aproximada. La tasa de escritura real dependerá de factores tales como el hecho de que exista una distribución uniforme de claves en DynamoDB Si encuentra que la operación de Hive suele superar el rendimiento aprovisionado o si el tráfico de escritura en directo se está demasiado limitado, entonces reduzca este valor por debajo de `0.5`. Si dispone de suficiente capacidad y desea una operación de Hive más rápida, defina este valor por encima de `0.5`. También puede suscribir en exceso definiendo este valor en 1,5 si cree que hay operaciones de entrada/salida sin utilizar disponibles o que estos son los datos iniciales cargados en la tabla y no hay tráfico activo todavía. |
| dynamodb.endpoint | Especifique el punto de conexión del servicio de DynamoDB. Para obtener más información sobre los puntos de conexión de DynamoDB disponibles, consulte [Regiones y puntos de conexión](https://docs.aws.amazon.com/general/latest/gr/rande.html#ddb_region). |
| dynamodb.max.map.tasks | Especifique el número máximo de tareas de asignación al leer datos desde DynamoDB. Este valor debe ser igual o superior a 1. |
| dynamodb.retry.duration | Especifique el número de minutos que utilizar como duración de tiempo de inactividad para volver a intentar ejecutar los comandos de Hive. Este valor debe ser un entero igual o superior a 0. El tiempo de espera predeterminado tiene una duración de dos minutos. |
Estas opciones se establecen utilizando el comando `SET` tal y como se muestra en el siguiente ejemplo.
```
SET dynamodb.throughput.read.percent=1.0;
INSERT OVERWRITE TABLE s3_export SELECT *
FROM hiveTableName;
```
# Ejemplo de comandos de Hive para la exportación, importación y consulta de datos en DynamoDB
Los siguientes ejemplos utilizan comandos de Hive para llevar a cabo operaciones como: exportación de datos a Amazon S3 o HDFS, importación de datos a DynamoDB, unión de tablas, consulta de tablas y mucho más.
Las operaciones en una tabla de Hive hacen referencia los datos almacenados en DynamoDB. Los comandos de Hive están sujetos a la configuración de rendimiento aprovisionado de la tabla de DynamoDB y los datos recuperados incluyen los datos escritos en la tabla de DynamoDB en el momento en que DynamoDB procesa la solicitud de la operación de Hive. Si el proceso de recuperación de datos tarda mucho tiempo, algunos datos devueltos por el comando Hive es posible que se hayan actualizado en DynamoDB desde que se inició el comando de Hive.
Los comandos de Hive `DROP TABLE` y `CREATE TABLE` solo actúan sobre la tablas locales en Hive y no crean ni borran tablas en DynamoDB. Si su consulta de Hive hace referencia a una tabla en DynamoDB, la tabla ya existe antes de ejecutar la consulta. Para obtener más información sobre la creación y eliminación de tablas en DynamoDB, consulte, [Uso de tablas en DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.html) en la *Guía para desarrolladores de Amazon DynamoDB*.
**nota**
Al mapear una tabla de Hive a una ubicación de Amazon S3, no la mapee a la ruta raíz del bucket, s3://amzn-s3-demo-bucket, ya que esto puede provocar errores cuando Hive escriba los datos en Amazon S3. En lugar de ello asigne la tabla a una subruta del bucket, s3://amzn-s3-demo-bucket/mypath.
## Exportación de datos desde DynamoDB
Puede utilizar Hive para exportar datos desde DynamoDB.
**Para exportar una tabla de DynamoDB a un bucket de Amazon S3**
+ Cree una tabla de Hive que haga referencia a los datos almacenados en DynamoDB. A continuación, puede llamar al comando INSERT OVERWRITE para escribir los datos en un directorio externo. En el siguiente ejemplo, *s3://amzn-s3-demo-bucket/path/subpath/* es una ruta válida en Amazon S3. Ajuste las columnas y tipos de datos en el comando CREATE para asignar los valores en su DynamoDB. Puede utilizar esto para crear un archivo de los datos de DynamoDB en 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;
```
**Para exportar una tabla de DynamoDB a un bucket de Amazon S3 utilizando del formato**
+ Cree una tabla externa que haga referencia a una ubicación en Amazon S3. Esto se muestra a continuación como s3\$1export. Durante la llamada CREATE, especifique el formato de fila para la tabla. A continuación, cuando utilice INSERT OVERWRITE para exportar datos desde DynamoDB a s3\$1export, los datos se escriben en el formato especificado. En el siguiente ejemplo, los datos se escriben como valores separados por comas (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;
```
**Para exportar una tabla de DynamoDB a un bucket de Amazon S3 sin especificar un mapeo de columnas**
+ Cree una tabla de Hive que haga referencia a los datos almacenados en DynamoDB. Esto es similar al ejemplo anterior, salvo que no tiene que especificar un mapeo de columnas. En la tabla debe tener exactamente una columna de tipo `map`. Si crea una tabla `EXTERNAL` en Amazon S3, puede llamar al comando `INSERT OVERWRITE` para escribir los datos de DynamoDB en Amazon S3. Puede utilizar esto para crear un archivo de los datos de DynamoDB en Amazon S3. Dado que no hay ningún mapeo de columnas, no puede consultar las tablas que se hayan exportado de esta manera. La exportación de datos sin especificar un mapeo de columna está disponible en Hive 0.8.1.5 o versiones posteriores, que es compatible con la AMI 2.2.*x* y posteriores de Amazon EMR.
```
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;
```
**Para exportar una tabla de DynamoDB a un bucket de Amazon S3 utilizando la compresión de datos**
+ Hive proporciona varios códecs de compresión que puede definir durante la sesión de Hive. Al hacerlo los datos exportados se comprimen en el formato especificado. En el siguiente ejemplo, se comprimen los archivos exportados mediante el Lempel-Ziv-Oberhumer algoritmo (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;
```
Los códecs de compresión disponibles son:
+ 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. BZip2Códec
+ org.apache.hadoop.io.compress. SnappyCodec
**Para exportar una tabla de DynamoDB a HDFS**
+ Utilice el siguiente comando Hive, donde *hdfs:///directoryName* hay una ruta HDFS válida y *hiveTableName* es una tabla de Hive que hace referencia a DynamoDB. Esta operación de exportación es más rápida que la exportación de una tabla de DynamoDB a Amazon S3 debido a que Hive 0.7.1.1 utiliza HDFS como paso intermedio al exportar datos a Amazon S3. El siguiente ejemplo muestra también cómo definir `dynamodb.throughput.read.percent` en 1,0 para aumentar la tasa de solicitud de lectura.
```
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;
```
También puede exportar datos a HDFS mediante el formato y la compresión que se muestran más arriba para la exportación a Amazon S3. Para hacerlo, solo tiene que sustituir el directorio de Amazon S3 en los ejemplos anteriores por un directorio HDFS.
**Para leer datos de caracteres UTF-8 no imprimibles en Hive**
+ Puede leer y escribir datos de caracteres UTF-8 no imprimibles con Hive utilizando la cláusula `STORED AS SEQUENCEFILE` al crear la tabla. A SequenceFile es el formato de archivo binario de Hadoop; debe usar Hadoop para leer este archivo. En el siguiente ejemplo se muestra cómo exportar datos desde DynamoDB a Amazon S3. Puede utilizar esta funcionalidad para controlar caracteres codificados con UTF-8 no imprimibles.
```
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;
```
## Importación de datos a DynamoDB
Al escribir datos en DynamoDB utilizando Hive, debe asegurarse de que el número de unidades de capacidad de escritura sea mayor que el número de asignadores en el clúster. Por ejemplo, los clústeres que se ejecutan en instancias de EC2 m1.xlarge producen 8 asignadores por instancia. En el caso de un clúster; que tiene 10 instancias, eso supondría un total de 80 asignadores. Si sus unidades de capacidad de escritura no son mayores que el número de asignadores en el clúster, la operación de escritura de Hive puede consumir todo el rendimiento de escritura o intentar consumir más rendimiento del aprovisionado. Para obtener más información sobre el número de mapeadores creado por cada tipo de instancia de EC2, consulte [Configurar Hadoop](emr-hadoop-config.md).
El número de mapeadores de Hadoop se controla mediante las divisiones de entrada. Si hay muy pocas divisiones, el comando de escritura podría no ser capaz de consumir todo el rendimiento de escritura disponible.
Si existe un elemento con la misma clave en la tabla de DynamoDB de destino, se sobrescribe. Si no existe ningún elemento con la clave en la tabla de DynamoDB de destino, el elemento se inserta.
**Para importar datos de Amazon S3 a DynamoDB**
+ Puede usar Amazon EMR (Amazon EMR) y Hive para escribir datos de Amazon S3 en 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;
```
**Para importar una tabla desde un bucket de Amazon S3 a DynamoDB sin especificar un mapeo de columnas**
+ Cree una tabla `EXTERNAL` que haga referencia a los datos almacenados en Amazon S3 que se exportaron previamente desde DynamoDB. Antes de la importación, asegúrese de que la tabla existe en DynamoDB y que tiene el mismo esquema de claves que la tabla de DynamoDB exportada anteriormente. Además, la tabla debe tener exactamente una columna de tipo `map`. Si, a continuación, crea una tabla de Hive que está vinculada a DynamoDB, puede llamar al comando `INSERT OVERWRITE` para escribir los datos de Amazon S3 en DynamoDB. Dado que no hay ningún mapeo de columnas, no puede consultar las tablas que se hayan importado de esta manera. La importación de datos sin especificar un mapeo de columna está disponible en Hive 0.8.1.5 o versiones posteriores, que es compatible con la AMI 2.2.3 y posteriores de Amazon EMR.
```
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;
```
**Para importar una tabla desde HDFS a DynamoDB**
+ Puede utilizar Amazon EMR y Hive para escribir datos de HDFS en 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;
```
## Consulta de datos en DynamoDB
Los siguientes ejemplos muestran las distintas formas en las que puede utilizar Amazon EMR para consultar los datos almacenados en DynamoDB.
**Para encontrar el mayor valor para una columna mapeada (`max`)**
+ Utilice los comandos de Hive como los siguientes. En el primer comando, la instrucción CREATE crea una tabla de Hive que hace referencia a los datos almacenados en DynamoDB. La instrucción SELECT utiliza a continuación dicha tabla para consultar los datos almacenados en DynamoDB. El siguiente ejemplo encuentra el mayor pedido realizado por un determinado cliente.
```
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;
```
**Para agregar datos utilizando la cláusula `GROUP BY`**
+ Puede utilizar la cláusula `GROUP BY` para recopilar datos a través de varios registros. Esto se utiliza a menudo con una función de agregación, como suma, recuento, mín o máx. En el siguiente ejemplo, se devuelve una lista de los mayores pedidos de clientes que han realizado más de tres pedidos.
```
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;
```
**Para unir dos tablas de DynamoDB**
+ El siguiente ejemplo asocia dos tablas de Hive a los datos almacenados en DynamoDB. A continuación, hace una llamada para unir estas dos tablas. La unión se calcula en el clúster y se devuelve. La unión no tiene lugar en DynamoDB. Este ejemplo devuelve una lista de los clientes y de sus compras para clientes que han realizado más de dos pedidos.
```
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;
```
**Para unir dos tablas desde orígenes diferentes**
+ En el siguiente ejemplo, Customer\$1S3 es una tabla de Hive que carga un archivo CSV almacenado en Amazon S3 y hive\$1purchases es una tabla que hace referencia a datos de DynamoDB. El siguiente ejemplo une datos de clientes almacenados en un archivo CSV en Amazon S3 con datos de pedidos almacenados en DynamoDB para devolver un conjunto de datos que representa los pedidos realizados por los clientes que tienen "Miller" en su nombre.
```
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%';
```
**nota**
En los ejemplos anteriores, se incluyeron las instrucciones CREATE TABLE en cada ejemplo para mayor claridad e integridad. Al ejecutar varias consultas o exportar operaciones en una tabla de Hive determinada, solo tiene que crear la tabla una vez, al principio de la sesión de Hive.
# Optimización de rendimiento para las operaciones de Amazon EMR en DynamoDB
Las operaciones de Amazon EMR en una tabla de DynamoDB cuentan como operaciones de lectura y están sujetas a la configuración de rendimiento aprovisionado de la tabla. Amazon EMR implementa su propia lógica para intentar balancear la carga en su tabla de DynamoDB para minimizar la posibilidad de sobrepasar el rendimiento aprovisionado. Al final de cada consulta de Hive, Amazon EMR devuelve información sobre el clúster; utilizado para procesar la consulta, incluyendo el número de veces que se ha superado el rendimiento aprovisionado. Puede utilizar esta información, así como CloudWatch las métricas sobre el rendimiento de DynamoDB, para gestionar mejor la carga de la tabla de DynamoDB en solicitudes posteriores.
Los siguientes factores influyen en el rendimiento de las consultas de Hive cuando se trabaja con tablas de DynamoDB.
## Unidades de capacidad de lectura aprovisionada
Cuando ejecuta consultas de Hive frente a una tabla de DynamoDB, debe asegurarse de que ha aprovisionado una cantidad suficiente de unidades de capacidad de lectura.
Por ejemplo, suponga que ha aprovisionado 100 unidades de capacidad de lectura para su tabla de DynamoDB. Esto le permite realizar 100 lecturas o 409 600 bytes, por segundo. Si dicha tabla contiene 20 GB de datos (21 474 836 480 bytes) y la consulta de Hive realiza un análisis completo de la tabla, puede calcular aproximadamente cuánto tardará la consulta en ejecutarse:
* 21 474 836 480 / 409 600 = 52 429 segundos = 14,56 horas *
La única forma de reducir el tiempo necesario, sería ajustar las unidades de capacidad de lectura en el origen de la tabla de DynamoDB. Agregar más nodos al clúster de Amazon EMR no servirá.
En la salida de Hive, el porcentaje completado se actualiza cuando finalizan uno o varios procesos de mapeador. Para una tabla de DynamoDB grande con un ajuste bajo de capacidad de lectura aprovisionada, la salida de porcentaje de finalización podría no actualizarse durante mucho tiempo; en el caso anterior, el trabajo aparecerá con un 0% completo durante varias horas. Para ver un estado más detallado del progreso del trabajo, acceda a la consola de Amazon EMR; podrá ver el estado de la tarea de asignador individual y estadísticas de lecturas de datos.
También puede iniciar sesión en la interfaz de usuario de Hadoop en el nodo principal y ver las estadísticas de Hadoop. Esto le muestra el estado de tareas de mapeo individuales y algunos datos de estadísticas de lectura. Para obtener más información, consulte [Interfaces web alojadas en el nodo maestro](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html) en la *Guía de administración de Amazon EMR*.
## Ajuste de porcentaje de lectura
De forma predeterminada, Amazon EMR administra la carga de solicitudes de la tabla de DynamoDB de acuerdo con el rendimiento aprovisionado actual. Sin embargo, cuando Amazon EMR devuelve información sobre su trabajo que incluye un elevado número de respuestas que superan el rendimiento aprovisionado, puede ajustar la tasa de lectura predeterminada mediante el parámetro `dynamodb.throughput.read.percent` cuando configure la tabla de Hive. Para obtener más información acerca cómo configurar el parámetro de porcentaje de lectura, consulte [Opciones de Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Configuración de porcentaje de escritura
De forma predeterminada, Amazon EMR administra la carga de solicitudes de la tabla de DynamoDB de acuerdo con el rendimiento aprovisionado actual. Sin embargo, cuando Amazon EMR devuelve información sobre su trabajo que incluye un elevado número de respuestas que superan el rendimiento aprovisionado, puede ajustar la tasa de escritura predeterminada mediante el parámetro `dynamodb.throughput.write.percent` cuando configure la tabla de Hive. Para obtener más información acerca cómo configurar el parámetro de porcentaje de escritura, consulte [Opciones de Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Configuración de duración de reintento
De forma predeterminada, Amazon EMR vuelve a ejecutar una consulta de Hive si no devuelve un resultado en el plazo de dos minutos, el intervalo de reintento predeterminado. Puede ajustar este intervalo configurando el parámetro `dynamodb.retry.duration` al ejecutar una consulta de Hive. Para obtener más información acerca cómo configurar el parámetro de porcentaje de escritura, consulte [Opciones de Hive](EMR_Interactive_Hive.md#EMR_Hive_Options).
## Número de tareas de asignación
Los daemons de asignador que lanza Hadoop para procesar las solicitudes para exportar y consultar datos almacenados en DynamoDB están limitados a una tasa de lectura máxima de 1 MiB por segundo para limitar la capacidad de lectura utilizada. Si dispone de rendimiento aprovisionado adicional en DynamoDB, puede mejorar el rendimiento de las operaciones de consulta y exportación de Hive aumentando el número de daemons de asignador. Para ello, puede aumentar el número de instancias de EC2 en su clúster; *o* aumentar el número de daemons de asignador que se ejecutan en cada instancia de EC2.
Puede aumentar el número de instancias de EC2 en un clúster; parando el clúster; actual y volviéndolo a lanzar con un número mayor de instancias de EC2. Especifique el número de instancias de EC2 en el cuadro de diálogo **Configurar instancias de EC2** si está lanzando el clúster desde la consola de Amazon EMR o con la opción `‑‑num-instances` si está lanzando el clúster desde la CLI.
El número de tareas de asignación que se ejecuta en una instancia depende del tipo de instancia de EC2. Para obtener más información sobre los tipos de instancia de EC2 y el número de mapeadores que proporciona cada una, consulte [Configuración de tareas](emr-hadoop-task-config.md). Allí encontrará una sección "Task Configuration" para cada una de las configuraciones admitidas.
Otra forma de aumentar el número de daemons de mapeador consiste en cambiar el parámetro de configuración `mapreduce.tasktracker.map.tasks.maximum` de Hadoop por un valor mayor. Esto tiene la ventaja de proporcionarle más mapeadores sin aumentar el número o el tamaño de instancias de EC2, lo que ahorra dinero. Un inconveniente es que si este valor se configura muy alto puede hacer que las instancias de EC2 de su clúster se queden sin memoria. Para configurar `mapreduce.tasktracker.map.tasks.maximum`, lance el clúster y especifique un valor para `mapreduce.tasktracker.map.tasks.maximum` como una propiedad de la clasificación de configuración mapred-site. Esto se muestra en el siguiente ejemplo. Para obtener más información, consulte [Configuración de aplicaciones](emr-configure-apps.md).
```
{
"configurations": [
{
"classification": "mapred-site",
"properties": {
"mapred.tasktracker.map.tasks.maximum": "10"
}
}
]
}
```
## Solicitudes de datos paralelas
Varias solicitudes de datos, ya sea de más de un usuario o más de una aplicación en una única tabla podría agotar el rendimiento de lectura aprovisionado y un dar lugar a un rendimiento lento.
## Duración del proceso
La consistencia de datos en DynamoDB depende del orden de las operaciones de lectura y escritura en cada nodo. Aunque haya una consulta de Hive en curso, otra aplicación podría cargar nuevos datos en la tabla de DynamoDB o modificar o eliminar datos existentes. En este caso, los resultados de la consulta de Hive podrían no reflejar los cambios realizados en los datos mientras se ejecutaba la consulta.
## Evitar un rendimiento excesivo
Al ejecutar consultas de Hive en DynamoDB, tenga cuidado de no superar el rendimiento aprovisionado, ya que esto agotará la capacidad necesaria para las llamadas de la aplicación a `DynamoDB::Get`. Para asegurarte de que esto no ocurra, deberías controlar periódicamente el volumen de lectura y la limitación de las llamadas a las aplicaciones `DynamoDB::Get` consultando los registros y monitorizando las métricas en Amazon. CloudWatch
## Tiempo de solicitud
Programar las consultas de Hive que acceden a una tabla de DynamoDB de modo que se lleven a cabo en aquellos momentos en que la demanda de esa tabla de DynamoDB es más baja, mejora el rendimiento. Por ejemplo, si la mayoría de los usuarios de la aplicación viven en San Francisco, podría elegir exportar los datos cada día a las 4:00 h PST, cuando la mayoría de los usuarios están durmiendo, y no actualizan los registros de la base de datos de DynamoDB.
## Tablas basadas en tiempo
Si los datos se organizan como una serie de tablas de DynamoDB basadas en tiempo como, por ejemplo, una tabla al día, puede exportar los datos cuando la tabla deje de estar activa. Puede utilizar esta técnica para crear copias de seguridad de los datos en Amazon S3 de manera continuada.
## Datos archivados
Si tiene previsto ejecutar muchas consultas de Hive en los datos almacenados en DynamoDB y la aplicación puede tolerar los datos archivados, es posible que desee exportar los datos a HDFS o a Amazon S3 y ejecutar las consultas de Hive en una copia de los datos en lugar de DynamoDB. Esto conserva las operaciones de lectura y el rendimiento aprovisionado.
# Kinesis
Los clústeres de Amazon EMR pueden leer y procesar las transmisiones de Amazon Kinesis directamente mediante herramientas conocidas del ecosistema de Hadoop, como Hive, Pig MapReduce, la API de streaming de Hadoop y Cascading. También puede unir los datos en tiempo real de Amazon Kinesis con los datos existentes en Amazon S3, Amazon DynamoDB y HDFS en un clúster en ejecución. Puede cargar datos directamente de Amazon EMR en Amazon S3 o en DynamoDB para actividades posteriores al procesamiento. Para obtener información sobre las características y los precios del servicio Amazon Kinesis, consulte la página de [Amazon Kinesis](https://aws.amazon.com//kinesis).
## ¿Qué puede hacer con la integración de Amazon EMR y Amazon Kinesis?
La integración entre Amazon EMR y Amazon Kinesis facilita en gran medida ciertas actividades, como, por ejemplo:
+ **Análisis de registros de streaming**: puede analizar registros web de streaming para generar una lista de los 10 tipos de error principales cada pocos minutos por región, navegador y dominio de acceso.
+ **Interacción con los clientes**: puede escribir consultas que unan datos de secuencias de clic desde Amazon Kinesis con información de campañas publicitarias almacenada en una tabla de DynamoDB para identificar las categorías de anuncios más eficaces que se muestran en sitios web particulares.
+ **Consultas interactivas ad-hoc**: puede cargar datos periódicamente desde flujos de Amazon Kinesis en HDFS y hacer que estén disponibles como tabla de Impala local para consultas analíticas interactivas rápidas.
## Análisis con puntos de control de flujos de Amazon Kinesis
Los usuarios pueden ejecutar análisis por lotes periódicos de flujos de Amazon Kinesis en lo que se denominan *iteraciones*. Dado que los registros de datos de los flujos de Amazon Kinesis se recuperan utilizando un número de secuencia, los límites de las iteraciones se definen por números de secuencia iniciales y finales que Amazon EMR almacena en una tabla de DynamoDB. Por ejemplo, cuando finaliza `iteration0`, almacena el número de secuencia final en DynamoDB para que cuando comience el trabajo `iteration1`, este pueda recuperar los datos siguientes del flujo. Este mapeo de iteraciones en los datos de flujo se denomina *creación de puntos de comprobación*. Para obtener más información, consulte [Conector de Kinesis](https://aws.amazon.com/elasticmapreduce/faqs/#kinesis-connector).
Si una iteración ha sido objeto de control y el trabajo no ha podido procesar una iteración, Amazon EMR intenta volver a procesar los registros de esa iteración.
La creación de puntos de comprobación es una característica que le permite:
+ Comenzar el procesamiento de datos después de un número de secuencia procesado por una consulta anterior que se ejecutó en el mismo flujo y nombre lógico
+ Volver a procesar el mismo lote de datos desde Kinesis que se procesó en una consulta anterior
Para habilitar la creación de puntos de comprobación, defina el parámetro `kinesis.checkpoint.enabled` en `true` en sus scripts. Además, configure los siguientes parámetros:
| Opción de configuración | Description (Descripción) |
| --- | --- |
| kinesis.checkpoint.metastore.table.name | Nombre de la tabla de DynamoDB donde se almacenará la información de los puntos de comprobación |
| kinesis.checkpoint.metastore.hash.key.name | Nombre de clave hash para la tabla de DynamoDB |
| kinesis.checkpoint.metastore.hash.range.name | Nombre de clave de rango para la tabla de DynamoDB |
| kinesis.checkpoint.logical.name | Un nombre lógico para el procesamiento actual |
| kinesis.checkpoint.iteration.no | Número de iteración para procesamiento asociado con el nombre lógico |
| kinesis.rerun.iteration.without.wait | Valor booleano que indica si una iteración que ha devuelto un error se puede volver a ejecutar sin necesidad de esperar un tiempo de espera; el valor predeterminado es false |
### Recomendaciones de IOPS aprovisionadas para tablas de Amazon DynamoDB
El conector de Amazon EMR para Amazon Kinesis utiliza la base de datos de DynamoDB como respaldo para los metadatos de puntos de comprobación. Debe crear una tabla en DynamoDB antes de consumir los datos de un flujo de Amazon Kinesis con un clúster de Amazon EMR en intervalos de puntos de comprobación. La tabla debe estar en la misma región que su clúster de Amazon EMR. A continuación, se ofrecen recomendaciones generales para el número de IOPS que debe aprovisionar para sus tablas de DynamoDB; `j` es el número máximo de trabajos de Hadoop (con diferente combinación de nombre lógico y número de iteración) que pueden ejecutarse de forma simultánea y `s` es el número máximo de particiones que procesará cualquier trabajo:
Para **Read Capacity Units (Unidades de capacidad de lectura)**: `j`\$1`s`/`5`
Para **Write Capacity Units (Unidades de capacidad de escritura)**: `j`\$1`s`
## Consideraciones sobre el rendimiento
El rendimiento de particiones de Amazon Kinesis es directamente proporcional al tamaño de instancia de los nodos en los clústeres Amazon EMR y al tamaño de los registros en el flujo. Le recomendamos que use instancias m5.xlarge o superiores en los nodos maestros y principales.
## Programación del análisis de Amazon Kinesis con Amazon EMR
Cuando analiza datos de un flujo Amazon Kinesis activo, limitado por los tiempos de espera y una duración máxima de las interacciones, es importante que ejecute el análisis con frecuencia para recopilar detalles del flujo con periodicidad. Hay varias formas de ejecutar estos scripts y consultas en intervalos periódicos. Para tareas recurrentes como estas, recomendamos usar AWS Data Pipeline . *Para obtener más información, consulte y en la Guía para desarrolladores. [AWS Data Pipeline PigActivity[AWS Data Pipeline HiveActivity](https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-hiveactivity.html)AWS Data Pipeline](https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-pigactivity.html)*
# Migración del conector de Spark Kinesis al SDK 2.x para Amazon EMR 7.0
El AWS SDK proporciona un amplio conjunto de APIs bibliotecas para interactuar con los servicios de computación AWS en la nube, como la administración de credenciales y la conexión a los servicios de S3 y Kinesis. El conector de Spark Kinesis se utiliza para consumir datos de Kinesis Data Streams, y los datos recibidos se transforman y procesan en el motor de ejecución de Spark. Actualmente, este conector se basa en la versión 1.x del AWS SDK y Kinesis-client-library (KCL).
Como parte de la migración del AWS SDK 2.x, el conector de Spark Kinesis también se actualizó en consecuencia para que funcione con el SDK 2.x. En la versión 7.0 de Amazon EMR, Spark contiene la actualización 2.x del SDK que aún no está disponible en la versión comunitaria de Apache Spark. Si utiliza el conector de Spark Kinesis de una versión anterior a la 7.0, debe migrar los códigos de las aplicaciones para que se ejecuten en el SDK 2.x antes de poder migrar a Amazon EMR 7.0.
## Guías de migración
En esta sección, se describen los pasos para migrar una aplicación al conector de Spark Kinesis actualizado. Incluye guías para migrar a la biblioteca de clientes de Kinesis (KCL) 2.x, proveedores de AWS credenciales y clientes de AWS servicios en el SDK 2.x. AWS Como referencia, también incluye un [WordCount](https://github.com/apache/spark/blob/v3.5.0/connector/kinesis-asl/src/main/scala/org/apache/spark/examples/streaming/KinesisWordCountASL.scala)programa de ejemplo que utiliza el conector Kinesis.
**Topics**
+ [Migración de KCL de la versión 1.x a la 2.x](#migrating-spark-kinesis-KCL-from-1.x-to-2.x)
+ [Migración de proveedores de AWS credenciales del SDK 1.x al AWS 2.x](#migrating-spark-kinesis-creds-from-1.x-to-2.x)
+ [Migración de clientes de AWS servicios del AWS SDK 1.x al 2.x](#migrating-spark-kinesis-service-from-1.x-to-2.x)
+ [Ejemplos de código para aplicaciones de transmisión](#migrating-spark-kinesis-streaming-examples)
+ [Consideraciones a la hora de utilizar el conector de Spark Kinesis actualizado](#migrating-spark-kinesis-considerations)
### Migración de KCL de la versión 1.x a la 2.x
+ **Nivel de métricas y dimensiones en `KinesisInputDStream`**
Al crear una instancia de `KinesisInputDStream`, puede controlar el nivel de métricas y las dimensiones para la secuencia. En el siguiente ejemplo, se muestra cómo se pueden personalizar estos parámetros con 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()
```
En KCL 2.x, estas opciones de configuración tienen nombres de paquete diferentes. Para migrar a 2.x:
1. Cambie las instrucciones de importación de `com.amazonaws.services.kinesis.clientlibrary.lib.worker.KinesisClientLibConfiguration` y `com.amazonaws.services.kinesis.metrics.interfaces.MetricsLevel` a `software.amazon.kinesis.metrics.MetricsLevel` y `software.amazon.kinesis.metrics.MetricsUtil` respectivamente.
```
// 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. Sustituya la línea `metricsEnabledDimensionsKinesisClientLibConfiguration.DEFAULT_METRICS_ENABLED_DIMENSIONS.asScala.toSet` por `metricsEnabledDimensionsSet(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME)`
A continuación, se presenta una versión actualizada de `KinesisInputDStream` con dimensiones y niveles de métricas personalizados:
```
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()
```
+ Función del controlador de mensajes en `KinesisInputDStream`
Al crear una instancia de `KinesisInputDStream`, también puede proporcionar una “función del controlador de mensajes” que tome un registro de Kinesis y devuelva un objeto genérico T, en caso de que desee utilizar otros datos incluidos en un registro, como la clave de partición.
En KCL 1.x, la firma de la función del controlador de mensajes es: `Record => T`, donde Record es `com.amazonaws.services.kinesis.model.Record`. En KCL 2.x, la firma del controlador se cambia a:`KinesisClientRecord => T`, where is. KinesisClientRecord `software.amazon.kinesis.retrieval.KinesisClientRecord`
A continuación, se muestra un ejemplo de cómo proporcionar un controlador de mensajes en 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)
```
Para migrar el controlador de mensajes:
1. Cambie la instrucción de importación de `com.amazonaws.services.kinesis.model.Record` a `software.amazon.kinesis.retrieval.KinesisClientRecord`.
```
// import com.amazonaws.services.kinesis.model.Record
import software.amazon.kinesis.retrieval.KinesisClientRecord
```
1. Actualice la firma del método del controlador de mensajes.
```
//def addFive(r: Record): Int = JavaUtils.bytesToString(r.getData).toInt + 5
def addFive = (r: KinesisClientRecord) => JavaUtils.bytesToString(r.data()).toInt + 5
```
A continuación, se presenta un ejemplo actualizado de cómo proporcionar el controlador de mensajes en 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)
```
Para obtener información sobre cómo migrar de KCL 1.x a KCL 2.x, consulte [Migración de consumidores de KCL 1.x a KCL 2.x](https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html).
### Migración de proveedores de AWS credenciales del SDK 1.x al AWS 2.x
Los proveedores de credenciales se utilizan para obtener AWS credenciales con las que interactuar. AWS Se han realizado varios cambios en las interfaces y clases relacionadas con los proveedores de credenciales en el SDK 2.x. Puede encontrar más información [aquí](https://github.com/aws/aws-sdk-java-v2/blob/master/docs/LaunchChangelog.md#122-client-credentials). El conector Spark Kinesis ha definido una interfaz (`org.apache.spark.streaming.kinesis.SparkAWSCredentials`) y clases de implementación que devuelven la versión 1.x de los proveedores de AWS credenciales. Estos proveedores de credenciales son necesarios para inicializar los clientes de Kinesis. Por ejemplo, si utiliza este método `SparkAWSCredentials.provider` en las aplicaciones, necesitará actualizar los códigos para utilizar la versión 2.x de los proveedores de credenciales. AWS
A continuación se muestra un ejemplo del uso de los proveedores de credenciales en AWS el 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")
```
**Para migrar a SDK 2.x:**
1. Cambie la instrucción de importación de `com.amazonaws.auth.AWSCredentialsProvider` a `software.amazon.awssdk.auth.credentials.AwsCredentialsProvider`
```
//import com.amazonaws.auth.AWSCredentialsProvider
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider
```
1. Actualice los códigos restantes que utilizan esta clase.
```
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")
```
### Migración de clientes de AWS servicios del AWS SDK 1.x al 2.x
AWS los clientes de servicio tienen diferentes nombres de paquete en la versión 2.x (es decir,`software.amazon.awssdk`), mientras que en el SDK 1.x los usa. `com.amazonaws` Para obtener más información sobre los cambios de los clientes, consulte [aquí](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-whats-different.html). Si utiliza estos clientes de servicio en los códigos, deberá realizar la migración correspondiente de los clientes.
A continuación, se muestra un ejemplo de cómo crear un cliente en SDK 1.x:
```
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
import com.amazonaws.services.dynamodbv2.document.DynamoDB
AmazonDynamoDB ddbClient = AmazonDynamoDBClientBuilder.defaultClient();
AmazonDynamoDBClient ddbClient = new AmazonDynamoDBClient();
```
**Para migrar a 2.x:**
1. Cambie las instrucciones de importación para los clientes de servicio. Tomemos como ejemplo los clientes de DynamoDB. Tendría que cambiar `com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient` o `com.amazonaws.services.dynamodbv2.document.DynamoDB` a `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. Actualización de los códigos que inicializan los clientes
```
// AmazonDynamoDB ddbClient = AmazonDynamoDBClientBuilder.defaultClient();
// AmazonDynamoDBClient ddbClient = new AmazonDynamoDBClient();
DynamoDbClient ddbClient = DynamoDbClient.create();
DynamoDbClient ddbClient = DynamoDbClient.builder().build();
```
Para obtener más información sobre la migración del AWS SDK de la versión 1.x a la 2.x, consulte [¿Qué diferencias hay entre el AWS SDK para Java](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-whats-different.html) 1.x y 2.x?
### Ejemplos de código para aplicaciones de transmisión
```
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()
}
}
```
### Consideraciones a la hora de utilizar el conector de Spark Kinesis actualizado
+ Si sus aplicaciones utilizan la `Kinesis-producer-library` con una versión del JDK inferior a la 11, es posible que encuentre excepciones como `java.lang.NoClassDefFoundError: javax/xml/bind/DatatypeConverter`. Esto sucede porque EMR 7.0 incluye JDK 17 de forma predeterminada y los módulos J2EE se han eliminado de las bibliotecas estándar desde Java 11 en adelante. Esto se puede solucionar si agrega la siguiente dependencia en el archivo pom. Reemplace la versión de biblioteca por la que crea conveniente.
```
javax.xml.bind
jaxb-api
${jaxb-api.version}
```
+ El jar del conector de Spark Kinesis se encuentra en esta ruta después de crear un clúster de EMR: `/usr/lib/spark/connector/lib/`
# S3 (s3-dist-cp) DistCp
Apache DistCp es una herramienta de código abierto que puede utilizar para copiar grandes cantidades de datos. *S3 DistCp* es similar a Amazon S3 DistCp, pero está optimizado para funcionar con él AWS, especialmente. El comando para S3 DistCp en Amazon EMR versión 4.0 y versiones posteriores es `s3-dist-cp` el que se agrega como un paso en un clúster o en la línea de comandos. Con S3DistCp, puede copiar de manera eficiente grandes cantidades de datos de Amazon S3 a HDFS, donde se pueden procesar en los siguientes pasos de su clúster de Amazon EMR. También puede usar S3 DistCp para copiar datos entre buckets de Amazon S3 o desde HDFS a Amazon S3. S3 DistCp es más escalable y eficiente para copiar en paralelo grandes cantidades de objetos entre depósitos y AWS cuentas.
Para ver comandos específicos que demuestran la flexibilidad de S3DistCP en situaciones reales, consulte [Siete consejos para usar 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/) en el blog Big Data. AWS
Por ejemplo DistCp, S3 DistCp suele MapReduce copiar de forma distribuida. Comparte las tareas de copia, tratamiento de errores, recuperación y elaboración de informes entre varios servidores. Para obtener más información sobre el proyecto de código DistCp abierto Apache, consulte la [DistCpguía](http://hadoop.apache.org/docs/stable/hadoop-distcp/DistCp.html) en la documentación de Apache Hadoop.
Si S3 no puede copiar algunos o todos los archivos especificados, DistCp se produce un error en el paso del clúster y devuelve un código de error distinto de cero. Si esto ocurre, S3 DistCp no limpia los archivos parcialmente copiados.
**importante**
S3 DistCp no admite nombres de bucket de Amazon S3 que contengan el carácter de subrayado.
S3 no DistCp admite la concatenación de archivos de Parquet. Úselo en su lugar. PySpark Para obtener más información, consulte [Concatenación de archivos Parquet en Amazon EMR](https://aws.amazon.com/premiumsupport/knowledge-center/emr-concatenate-parquet-files/).
Para evitar errores de copia al utilizar S3DistCP para copiar un único archivo (en lugar de un directorio) de S3 a HDFS, utilice Amazon EMR versión 5.33.0 o posterior, o Amazon EMR versión 6.3.0 o posterior.
## DistCp Opciones de S3
Aunque es similar a DistCp, S3 DistCp admite un conjunto diferente de opciones para cambiar la forma en que copia y comprime los datos.
Al llamar a S3DistCp, puede especificar las opciones que se describen en la siguiente tabla. Las opciones se añaden al paso utilizando la lista de argumentos. En la siguiente tabla se muestran ejemplos de los DistCp argumentos de S3.
| Opción | Description (Descripción) | Obligatorio |
| --- | --- | --- |
| ‑‑src=LOCATION | Ubicación de los datos que copiar. Puede ser una ubicación de HDFS o Amazon S3. Ejemplo: `‑‑src=s3://amzn-s3-demo-bucket/logs/j-3GYXXXXXX9IOJ/node` S3 DistCp no admite nombres de bucket de Amazon S3 que contengan el carácter de subrayado. | Sí |
| ‑‑dest=LOCATION | Destino de los datos. Puede ser una ubicación de HDFS o Amazon S3. Ejemplo: `‑‑dest=hdfs:///output` S3 DistCp no admite nombres de bucket de Amazon S3 que contengan el carácter de subrayado. | Sí |
| ‑‑srcPattern=PATTERN | Una [expresión regular](http://en.wikipedia.org/wiki/Regular_expression) que filtra la operación de copia para un subconjunto de los datos de `‑‑src`. Si no se especifica `‑‑srcPattern` ni `‑‑groupBy`, todos los datos de `‑‑src` se copian en `‑‑dest`. Si el argumento de expresión regular contiene caracteres especiales, como un asterisco (\$1), la expresión regular o toda la cadena `‑‑args` debe estar escrita entre comillas ('). Ejemplo: `‑‑srcPattern=.*daemons.*-hadoop-.*` | No |
| ‑‑groupBy=PATTERN | [Expresión regular](http://en.wikipedia.org/wiki/Regular_expression) que hace que S3 DistCp concatene los archivos que coinciden con la expresión. Por ejemplo, podría utilizar esta opción para combinar todos los archivos de registro escritos en una hora en un solo archivo. El nombre de archivo concatenado es el valor asignado por la expresión regular para la agrupación. Los paréntesis indican cómo deberían agruparse los archivos, todos los elementos que coincidan con la instrucción parentética se combinan en un único archivo de salida. Si la expresión regular no incluye una sentencia entre paréntesis, el clúster falla en el DistCp paso S3 y devuelve un error. Si el argumento de expresión regular contiene caracteres especiales, como un asterisco (\$1), la expresión regular o toda la cadena `‑‑args` debe estar escrita entre comillas ('). Cuando se especifica `‑‑groupBy`, solo se copian los archivos que coinciden con el patrón especificado. No tiene que especificar `‑‑groupBy` y `‑‑srcPattern` al mismo tiempo. Ejemplo: `‑‑groupBy=.*subnetid.*([0-9]+-[0-9]+-[0-9]+-[0-9]+).*` | No |
| ‑‑targetSize=SIZE | El tamaño, en mebibytes (MiB), de los archivos a crear en función de la opción `‑‑groupBy`. Este valor debe ser un número entero. Cuando `‑‑targetSize` se establece, S3 DistCp intenta igualar este tamaño; el tamaño real de los archivos copiados puede ser mayor o menor que este valor. Los trabajos se agregan en función del tamaño del archivo de datos, por lo que es posible que el tamaño del archivo de destino coincida con el tamaño del archivo de datos de origen. Si los archivos concatenados por `‑‑groupBy` son mayores que el valor de `‑‑targetSize`, se dividen en archivos parciales y se nombran de forma secuencial con un valor numérico junto al final. Por ejemplo, un archivo concatenado en `myfile.gz` se desglosaría en partes como: `myfile0.gz`, `myfile1.gz`, etc. Ejemplo: `‑‑targetSize=2` | No |
| ‑‑appendToLastFile | Especifica el comportamiento de S3 DistCp al copiar archivos de Amazon S3 a HDFS que ya están presentes. Añade los nuevos datos de archivo a archivos existentes. Si utiliza `‑‑appendToLastFile` con `‑‑groupBy`, se añaden nuevos datos a los archivos que coincidirán con los mismos grupos. Esta opción también respeta el comportamiento `‑‑targetSize` cuando se utiliza con `‑‑groupBy.` | No |
| ‑‑outputCodec=CODEC | Especifica el códec de compresión que utilizar para los archivos copiados. Esto puede tomar los valores: `gzip`, `gz`, `lzo`, `snappy` o bien `none`. Puede utilizar esta opción, por ejemplo, para convertir archivos de entrada comprimidos con Gzip en archivos de salida con compresión LZO o para descomprimir los archivos como parte de la operación de copia. Si elige un códec de salida, el nombre de archivo se agregará con la extensión adecuada (por ejemplo, `gz` y `gzip`, la extensión es `.gz`). Si no especifica un valor para `‑‑outputCodec`, los archivos se copian sin cambios en su compresión. Ejemplo: `‑‑outputCodec=lzo` | No |
| ‑‑s3ServerSideEncryption | Garantiza que los datos de destino se transfieran mediante SSL y se cifren automáticamente en Amazon S3 mediante una clave AWS del lado del servicio. Al recuperar datos mediante S3DistCp, los objetos se descifran automáticamente. Si intenta copiar un objeto sin cifrar en un bucket de Amazon S3 con cifrado requerido, la operación produce un error. Para obtener más información, consulte el tema relacionado con el [uso del cifrado de datos](https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingEncryption.html). Ejemplo: `‑‑s3ServerSideEncryption` | No |
| ‑‑deleteOnSuccess | Si la operación de copia se realiza correctamente, esta opción hace DistCp que S3 elimine los archivos copiados de la ubicación de origen. Esto resulta útil si está copiando archivos de salida, como archivos de registro, desde una ubicación a otra como una tarea programada y no desea copiar los mismos archivos dos veces. Ejemplo: `‑‑deleteOnSuccess` | No |
| ‑‑disableMultipartUpload | Deshabilita el uso de la carga multiparte. Ejemplo: `‑‑disableMultipartUpload` | No |
| ‑‑multipartUploadChunkSize=SIZE | El tamaño, en MiB, de cada parte de una carga multiparte de Amazon S3. S3 DistCp utiliza la carga multiparte cuando copia datos de un tamaño superior al`multipartUploadChunkSize`. Para mejorar el rendimiento de los trabajos, puede aumentar el tamaño de cada parte. El tamaño predeterminado es 128 MB. Ejemplo: `‑‑multipartUploadChunkSize=1000` | No |
| ‑‑numberFiles | Añade archivos de salida con los números secuenciales. El recuento comienza en 0 a menos que se especifique un valor diferente mediante `‑‑startingIndex`. Ejemplo: `‑‑numberFiles` | No |
| ‑‑startingIndex=INDEX | Utilizado con `‑‑numberFiles` para especificar el primer número de la secuencia. Ejemplo: `‑‑startingIndex=1` | No |
| ‑‑outputManifest=FILENAME | Crea un archivo de texto, comprimido con Gzip, que contiene una lista de todos los archivos copiados por S3. DistCp Ejemplo: `‑‑outputManifest=manifest-1.gz` | No |
| ‑‑previousManifest=PATH | Lee un archivo de manifiesto que se creó durante una llamada anterior a S3 DistCp con la `‑‑outputManifest` marca. Cuando el `‑‑previousManifest` indicador está activado, S3 DistCp excluye de la operación de copia los archivos que figuran en el manifiesto. Si se especifica `‑‑outputManifest` junto con los archivos `‑‑previousManifest`, incluidos en el manifiesto anterior también aparecen en el nuevo archivo de manifiesto, aunque los archivos no se copiarán. Ejemplo: `‑‑previousManifest=/usr/bin/manifest-1.gz` | No |
| ‑‑requirePreviousManifest | Requiere un manifiesto previo creado durante una llamada anterior a S3DistCp. Si se define como false, no se genera ningún error cuando no se especifica un manifiesto anterior. El valor predeterminado es true. | No |
| ‑‑copyFromManifest | Invierte el comportamiento de `‑‑previousManifest` hacer que S3 DistCp utilice el archivo de manifiesto especificado como una lista de archivos para copiar, en lugar de una lista de archivos para excluirlos de la copia. Ejemplo: `‑‑copyFromManifest ‑‑previousManifest=/usr/bin/manifest-1.gz` | No |
| ‑‑s3Endpoint=ENDPOINT | Especifica el punto de conexión de Amazon S3 que utilizar a la hora de cargar un archivo. Esta opción establece el punto de enlace para el origen y el destino. Si no se define, el punto de enlace predeterminado es `s3.amazonaws.com`. Para ver una lista de los puntos de conexión de Amazon S3, consulte [Regiones y puntos de conexión](https://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region). Ejemplo: `‑‑s3Endpoint=s3.eu-west-1.amazonaws.com` | No |
| ‑‑storageClass=CLASS | La clase de almacenamiento que utilizar cuando el destino es Amazon S3. Los valores válidos son STANDARD y REDUCED\$1REDUNDANCY. Si no se especifica esta opción, S3 DistCp intentará conservar la clase de almacenamiento. Ejemplo: `‑‑storageClass=STANDARD` | No |
| ‑‑srcPrefixesFile=PATH | un archivo de texto en Amazon S3 (s3://), HDFS (hdfs:///) o sistema de archivos local (file:/) que contiene una lista de prefijos `src`, un prefijo por línea. Si `srcPrefixesFile` se proporciona, S3 no DistCp mostrará la ruta src. En lugar de ello, genera una lista de origen como resultado combinado de enumerar todos los prefijos especificados en este archivo. La ruta relativa en comparación con la ruta src, en lugar de estos prefijos, se utilizará para generar las rutas de destino. Si también se especifica `srcPattern`, se aplicará a los resultados de lista combinada de los prefijos de origen para filtrar aún más la entrada. Si se utiliza `copyFromManifest`, lo objetos en el manifiesto se copiarán y no se tendrá en cuenta `srcPrefixesFile`. Ejemplo: `‑‑srcPrefixesFile=PATH` | No |
Además de las opciones anteriores, S3 DistCp implementa la [interfaz Tool](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/util/Tool.html), lo que significa que admite las opciones genéricas.
## Añadir S3 DistCp como un paso de un clúster
Para llamar a S3, DistCp agréguelo como un paso en su clúster. Los pasos se pueden añadir a un clúster en el momento de su lanzamiento o a un clúster en ejecución mediante la consola, la CLI o la API. En los siguientes ejemplos, se muestra cómo añadir un DistCp paso de S3 a un clúster en ejecución. Para obtener más información sobre cómo agregar pasos a un clúster, consulte [Enviar trabajo a un clúster](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-work-with-steps.html) en la *Guía de administración de Amazon EMR*.
**Para añadir un DistCp paso S3 a un clúster en ejecución mediante el AWS CLI**
Para obtener más información sobre el uso de los comandos de Amazon EMR en AWS CLI, consulte la Referencia de [AWS CLI comandos](https://docs.aws.amazon.com/cli/latest/reference/emr).
+ Para añadir un paso a un clúster que llame a S3DistCp, pase los parámetros que especifican cómo S3 DistCp debe realizar la operación de copia como argumentos.
El siguiente ejemplo copia registros de daemon desde Amazon S3 a `hdfs:///output`. En el comando siguiente:
+ `‑‑cluster-id` especifica el clúster
+ `Jar`es la ubicación del archivo DistCp JAR de S3. Para ver un ejemplo de cómo ejecutar un comando en un clúster mediante command-runner.jar, consulte [Envío de un paso JAR personalizado para ejecutar un script o un comando](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-commandrunner.html#emr-commandrunner-examples).
+ `Args`es una lista separada por comas de los pares nombre-valor de las opciones que se van a pasar a S3. DistCp Para ver una lista completa de las opciones disponibles, consulte [DistCp Opciones de S3](#UsingEMR_s3distcp.options).
Para añadir un paso de DistCp copia de S3 a un clúster en ejecución, coloque lo siguiente en un archivo JSON guardado en Amazon S3 o en su sistema de archivos local, como en este `myStep.json` ejemplo. *j-3GYXXXXXX9IOK*Sustitúyalo por tu ID de clúster y *amzn-s3-demo-bucket* sustitúyelo por el nombre de tu bucket de 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 los archivos de registro desde Amazon S3 a HDFS**
Este ejemplo también muestra cómo copiar archivos de registro almacenados en un bucket de Amazon S3 en HDFS agregando un paso a un clúster en ejecución. En este ejemplo, la opción `‑‑srcPattern` se utiliza para limitar los datos copiados en los registros del daemon.
Para copiar los archivos de registro desde Amazon S3 en HDFS mediante la opción `‑‑srcPattern`, coloque lo siguiente en un archivo JSON guardado en Amazon S3 o en el sistema de archivos local, como `myStep.json` en este ejemplo. *j-3GYXXXXXX9IOK*Sustitúyalo por tu ID de clúster y *amzn-s3-demo-bucket* sustitúyelo por el nombre de tu bucket de 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"
}
]
```
## Limpiando después de DistCp trabajos fallidos en S3
Si S3 DistCp no puede copiar algunos o todos los archivos especificados, el comando o el paso del clúster fallará y devolverá un código de error distinto de cero. Si esto ocurre, S3 DistCp no limpia los archivos parcialmente copiados. Deben eliminarse manualmente.
Los archivos copiados parcialmente se guardan en el `tmp` directorio HDFS, en subdirectorios, con el identificador único del trabajo de S3DistCp . Encontrará este ID en la salida estándar del trabajo.
Por ejemplo, para un DistCp trabajo de S3 con el ID`4b1c37bb-91af-4391-aaf8-46a6067085a6`, puede conectarse al nodo principal del clúster y ejecutar el siguiente comando para ver los archivos de salida asociados al trabajo.
```
hdfs dfs -ls /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6/output
```
El comando devuelve una lista de archivos similar a la siguiente:
```
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
```
A continuación, puede ejecutar el siguiente comando para eliminar el directorio y todo el contenido.
```
hdfs dfs rm -rf /tmp/4b1c37bb-91af-4391-aaf8-46a6067085a6
```