# Export, import, query, and join tables in DynamoDB using Amazon EMR **Note** The Amazon EMR-DynamoDB Connector is open-sourced on GitHub. For more information, see [https://github.com/awslabs/emr-dynamodb-connector](https://github.com/awslabs/emr-dynamodb-connector). DynamoDB is a fully managed NoSQL database service that provides fast and predictable performance with seamless scalability. Developers can create a database table and grow its request traffic or storage without limit. DynamoDB automatically spreads the data and traffic for the table over a sufficient number of servers to handle the request capacity specified by the customer and the amount of data stored, while maintaining consistent, fast performance. Using Amazon EMR and Hive you can quickly and efficiently process large amounts of data, such as data stored in DynamoDB. For more information about DynamoDB, see [Amazon DynamoDB Developer Guide](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/). Apache Hive is a software layer that you can use to query map reduce clusters using a simplified, SQL-like query language called HiveQL. It runs on top of the Hadoop architecture. For more information about Hive and HiveQL, go to the [HiveQL language manual](https://cwiki.apache.org/confluence/display/Hive/LanguageManual). For more information about Hive and Amazon EMR, see [Apache Hive](emr-hive.md) . You can use Amazon EMR with a customized version of Hive that includes connectivity to DynamoDB to perform operations on data stored in DynamoDB: + Loading DynamoDB data into the Hadoop Distributed File System (HDFS) and using it as input into an Amazon EMR cluster. + Querying live DynamoDB data using SQL-like statements (HiveQL). + Joining data stored in DynamoDB and exporting it or querying against the joined data. + Exporting data stored in DynamoDB to Amazon S3. + Importing data stored in Amazon S3 to DynamoDB. **Note** The Amazon EMR-DynamoDB Connector does not support clusters configured to use [Kerberos authentication](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-kerberos.html). To perform each of the following tasks, you'll launch an Amazon EMR cluster, specify the location of the data in DynamoDB, and issue Hive commands to manipulate the data in DynamoDB. There are several ways to launch an Amazon EMR cluster: you can use the Amazon EMR console, the command line interface (CLI), or you can program your cluster using an AWS SDK or the Amazon EMR API. You can also choose whether to run a Hive cluster interactively or from a script. In this section, we will show you how to launch an interactive Hive cluster from the Amazon EMR console and the CLI. Using Hive interactively is a great way to test query performance and tune your application. After you have established a set of Hive commands that will run on a regular basis, consider creating a Hive script that Amazon EMR can run for you. **Warning** Amazon EMR read or write operations on an DynamoDB table count against your established provisioned throughput, potentially increasing the frequency of provisioned throughput exceptions. For large requests, Amazon EMR implements retries with exponential backoff to manage the request load on the DynamoDB table. Running Amazon EMR jobs concurrently with other traffic may cause you to exceed the allocated provisioned throughput level. You can monitor this by checking the **ThrottleRequests** metric in Amazon CloudWatch. If the request load is too high, you can relaunch the cluster and set the [Read percent setting](EMR_Hive_Optimizing.md#ReadPercent) or [Write percent setting](EMR_Hive_Optimizing.md#WritePercent) to a lower value to throttle the Amazon EMR operations. For information about DynamoDB throughput settings, see [Provisioned throughput](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithDDTables.html#ProvisionedThroughput). If a table is configured for [On-Demand mode](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.ReadWriteCapacityMode.html#HowItWorks.OnDemand), you should change the table back to provisioned mode before running an export or import operation. Pipelines need a throughput ratio in order to calculate resources to use from a DynamoDBtable. On-demand mode removes provisioned throughput. To provision throughput capacity, you can use Amazon CloudWatch Events metrics to evaluate the aggregate throughput that a table has used. **Topics** + [Set up a Hive table to run Hive commands](EMR_Interactive_Hive.md) + [Hive command examples for exporting, importing, and querying data in DynamoDB](EMR_Hive_Commands.md) + [Optimizing performance for Amazon EMR operations in DynamoDB](EMR_Hive_Optimizing.md) # Set up a Hive table to run Hive commands Apache Hive is a data warehouse application you can use to query data contained in Amazon EMR clusters using a SQL-like language. For more information about Hive, see [http://hive.apache.org/](http://hive.apache.org/). The following procedure assumes you have already created a cluster and specified an Amazon EC2 key pair. To learn how to get started creating clusters, see [Getting started with Amazon EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-gs) in the *Amazon EMR Management Guide*. ## Configure Hive to use MapReduce When you use Hive on Amazon EMR to query DynamoDB tables, errors can occur if Hive uses the default execution engine, Tez. For this reason, when you create a cluster with Hive that integrates with DynamoDB as described in this section, we recommend that you use a configuration classification that sets Hive to use MapReduce. For more information, see [Configure applications](emr-configure-apps.md). The following snippet shows the configuration classification and property to use to set MapReduce as the execution engine for Hive: ``` [ { "Classification": "hive-site", "Properties": { "hive.execution.engine": "mr" } } ] ``` **To run Hive commands interactively** 1. Connect to the master node. For more information, see [Connect to the master node using SSH](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html) in the *Amazon EMR Management Guide*. 1. At the command prompt for the current master node, type `hive`. You should see a hive prompt: `hive>` 1. Enter a Hive command that maps a table in the Hive application to the data in DynamoDB. This table acts as a reference to the data stored in Amazon DynamoDB; the data is not stored locally in Hive and any queries using this table run against the live data in DynamoDB, consuming the table's read or write capacity every time a command is run. If you expect to run multiple Hive commands against the same dataset, consider exporting it first. The following shows the syntax for mapping a Hive table to a DynamoDB table. ``` 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..."); ``` When you create a table in Hive from DynamoDB, you must create it as an external table using the keyword `EXTERNAL`. The difference between external and internal tables is that the data in internal tables is deleted when an internal table is dropped. This is not the desired behavior when connected to Amazon DynamoDB, and thus only external tables are supported. For example, the following Hive command creates a table named *hivetable1* in Hive that references the DynamoDB table named *dynamodbtable1*. The DynamoDB table *dynamodbtable1* has a hash-and-range primary key schema. The hash key element is `name` (string type), the range key element is `year` (numeric type), and each item has an attribute value for `holidays` (string set type). ``` 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"); ``` Line 1 uses the HiveQL `CREATE EXTERNAL TABLE` statement. For *hivetable1*, you need to establish a column for each attribute name-value pair in the DynamoDB table, and provide the data type. These values are not case-sensitive, and you can give the columns any name (except reserved words). Line 2 uses the `STORED BY` statement. The value of `STORED BY` is the name of the class that handles the connection between Hive and DynamoDB. It should be set to `'org.apache.hadoop.hive.dynamodb.DynamoDBStorageHandler'`. Line 3 uses the `TBLPROPERTIES` statement to associate "hivetable1" with the correct table and schema in DynamoDB. Provide `TBLPROPERTIES` with values for the `dynamodb.table.name` parameter and `dynamodb.column.mapping` parameter. These values *are* case-sensitive. **Note** All DynamoDB attribute names for the table must have corresponding columns in the Hive table. Depending on your Amazon EMR version, the following scenarios occur if the one-to-one mapping does not exist: On Amazon EMR version 5.27.0 and later, the connector has validations that ensure a one-to-one mapping between DynamoDB attribute names and columns in the Hive table. An error will occur if the one-to-one mapping does not exist. On Amazon EMR version 5.26.0 and earlier, the Hive table won't contain the name-value pair from DynamoDB. If you do not map the DynamoDB primary key attributes, Hive generates an error. If you do not map a non-primary key attribute, no error is generated, but you won't see the data in the Hive table. If the data types do not match, the value is null. Then you can start running Hive operations on *hivetable1*. Queries run against *hivetable1* are internally run against the DynamoDB table *dynamodbtable1* of your DynamoDB account, consuming read or write units with each execution. When you run Hive queries against a DynamoDB table, you need to ensure that you have provisioned a sufficient amount of read capacity units. For example, suppose that you have provisioned 100 units of read capacity for your DynamoDB table. This will let you perform 100 reads, or 409,600 bytes, per second. If that table contains 20GB of data (21,474,836,480 bytes), and your Hive query performs a full table scan, you can estimate how long the query will take to run: * 21,474,836,480 / 409,600 = 52,429 seconds = 14.56 hours * The only way to decrease the time required would be to adjust the read capacity units on the source DynamoDB table. Adding more Amazon EMR nodes will not help. In the Hive output, the completion percentage is updated when one or more mapper processes are finished. For a large DynamoDB table with a low provisioned read capacity setting, the completion percentage output might not be updated for a long time; in the case above, the job will appear to be 0% complete for several hours. For more detailed status on your job's progress, go to the Amazon EMR console; you will be able to view the individual mapper task status, and statistics for data reads. You can also log on to Hadoop interface on the master node and see the Hadoop statistics. This will show you the individual map task status and some data read statistics. For more information, see the following topics: + [Web interfaces hosted on the master node](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html) + [View the Hadoop web interfaces](https://docs.aws.amazon.com/emr/latest/ManagementGuide/UsingtheHadoopUserInterface.html) For more information about sample HiveQL statements to perform tasks such as exporting or importing data from DynamoDB and joining tables, see [Hive command examples for exporting, importing, and querying data in DynamoDB](EMR_Hive_Commands.md). **To cancel a Hive request** When you execute a Hive query, the initial response from the server includes the command to cancel the request. To cancel the request at any time in the process, use the **Kill Command** from the server response. 1. Enter `Ctrl+C` to exit the command line client. 1. At the shell prompt, enter the **Kill Command** from the initial server response to your request. Alternatively, you can run the following command from the command line of the master node to kill the Hadoop job, where *job-id* is the identifier of the Hadoop job and can be retrieved from the Hadoop user interface. ``` hadoop job -kill job-id ``` ## Data types for Hive and DynamoDB The following table shows the available Hive data types, the default DynamoDB type that they correspond to, and the alternate DynamoDB types that they can also map to. | Hive type | Default DynamoDB type | Alternate DynamoDB type(s) | | --- | --- | --- | | string | string (S) | | | bigint or double | number (N) | | | binary | binary (B) | | | boolean | boolean (BOOL) | | | array | list (L) | number set (NS), string set (SS), or binary set (BS) | | map | item | map (M) | | map | map (M) | | | | null (NULL) | | If you want to write your Hive data as a corresponding alternate DynamoDB type, or if your DynamoDB data contains attribute values of an alternate DynamoDB type, you can specify the column and the DynamoDB type with the `dynamodb.type.mapping` parameter. The following example shows the syntax for specifying an alternate type mapping. ``` 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"); ``` The type mapping parameter is optional, and only has to be specified for the columns that use alternate types. For example, the following Hive command creates a table named `hivetable2` that references the DynamoDB table `dynamodbtable2`. It is similar to `hivetable1`, except that it maps the `col3` column to the string set (SS) type. ``` 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"); ``` In Hive, `hivetable1` and `hivetable2` are identical. However, when data from those tables are written to their corresponding DynamoDB tables, `dynamodbtable1` will contain lists, while `dynamodbtable2` will contain string sets. If you want to write Hive `null` values as attributes of DynamoDB `null` type, you can do so with the `dynamodb.null.serialization` parameter. The following example shows the syntax for specifying `null` serialization. ``` 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"); ``` The null serialization parameter is optional, and is set to `false` if not specified. Note that DynamoDB `null` attributes are read as `null` values in Hive regardless of the parameter setting. Hive collections with `null` values can be written to DynamoDB only if the null serialization parameter is specified as `true`. Otherwise, a Hive error occurs. The bigint type in Hive is the same as the Java long type, and the Hive double type is the same as the Java double type in terms of precision. This means that if you have numeric data stored in DynamoDB that has precision higher than is available in the Hive datatypes, using Hive to export, import, or reference the DynamoDB data could lead to a loss in precision or a failure of the Hive query. Exports of the binary type from DynamoDB to Amazon Simple Storage Service (Amazon S3) or HDFS are stored as a Base64-encoded string. If you are importing data from Amazon S3 or HDFS into the DynamoDB binary type, it should be encoded as a Base64 string. ## Hive options You can set the following Hive options to manage the transfer of data out of Amazon DynamoDB. These options only persist for the current Hive session. If you close the Hive command prompt and reopen it later on the cluster, these settings will have returned to the default values. | Hive options | Description | | --- | --- | | dynamodb.throughput.read.percent | Set the rate of read operations to keep your DynamoDB provisioned throughput rate in the allocated range for your table. The value is between `0.1` and `1.5`, inclusively. The value of 0.5 is the default read rate, which means that Hive will attempt to consume half of the read provisioned throughout resources in the table. Increasing this value above 0.5 increases the read request rate. Decreasing it below 0.5 decreases the read request rate. This read rate is approximate. The actual read rate will depend on factors such as whether there is a uniform distribution of keys in DynamoDB. If you find your provisioned throughput is frequently exceeded by the Hive operation, or if live read traffic is being throttled too much, then reduce this value below `0.5`. If you have enough capacity and want a faster Hive operation, set this value above `0.5`. You can also oversubscribe by setting it up to 1.5 if you believe there are unused input/output operations available. | | dynamodb.throughput.write.percent | Set the rate of write operations to keep your DynamoDB provisioned throughput rate in the allocated range for your table. The value is between `0.1` and `1.5`, inclusively. The value of 0.5 is the default write rate, which means that Hive will attempt to consume half of the write provisioned throughout resources in the table. Increasing this value above 0.5 increases the write request rate. Decreasing it below 0.5 decreases the write request rate. This write rate is approximate. The actual write rate will depend on factors such as whether there is a uniform distribution of keys in DynamoDB If you find your provisioned throughput is frequently exceeded by the Hive operation, or if live write traffic is being throttled too much, then reduce this value below `0.5`. If you have enough capacity and want a faster Hive operation, set this value above `0.5`. You can also oversubscribe by setting it up to 1.5 if you believe there are unused input/output operations available or this is the initial data upload to the table and there is no live traffic yet. | | dynamodb.endpoint | Specify the endpoint for the DynamoDB service. For more information about the available DynamoDB endpoints, see [Regions and endpoints](https://docs.aws.amazon.com/general/latest/gr/rande.html#ddb_region). | | dynamodb.max.map.tasks | Specify the maximum number of map tasks when reading data from DynamoDB. This value must be equal to or greater than 1. | | dynamodb.retry.duration | Specify the number of minutes to use as the timeout duration for retrying Hive commands. This value must be an integer equal to or greater than 0. The default timeout duration is two minutes. | These options are set using the `SET` command as shown in the following example. ``` SET dynamodb.throughput.read.percent=1.0; INSERT OVERWRITE TABLE s3_export SELECT * FROM hiveTableName; ``` # Hive command examples for exporting, importing, and querying data in DynamoDB The following examples use Hive commands to perform operations such as exporting data to Amazon S3 or HDFS, importing data to DynamoDB, joining tables, querying tables, and more. Operations on a Hive table reference data stored in DynamoDB. Hive commands are subject to the DynamoDB table's provisioned throughput settings, and the data retrieved includes the data written to the DynamoDB table at the time the Hive operation request is processed by DynamoDB. If the data retrieval process takes a long time, some data returned by the Hive command may have been updated in DynamoDB since the Hive command began. Hive commands `DROP TABLE` and `CREATE TABLE` only act on the local tables in Hive and do not create or drop tables in DynamoDB. If your Hive query references a table in DynamoDB, that table must already exist before you run the query. For more information about creating and deleting tables in DynamoDB, see [Working with tables in DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.html) in the *Amazon DynamoDB Developer Guide*. **Note** When you map a Hive table to a location in Amazon S3, do not map it to the root path of the bucket, s3://amzn-s3-demo-bucket, as this may cause errors when Hive writes the data to Amazon S3. Instead map the table to a subpath of the bucket, s3://amzn-s3-demo-bucket/mypath. ## Exporting data from DynamoDB You can use Hive to export data from DynamoDB. **To export a DynamoDB table to an Amazon S3 bucket** + Create a Hive table that references data stored in DynamoDB. Then you can call the INSERT OVERWRITE command to write the data to an external directory. In the following example, *s3://amzn-s3-demo-bucket/path/subpath/* is a valid path in Amazon S3. Adjust the columns and datatypes in the CREATE command to match the values in your DynamoDB. You can use this to create an archive of your DynamoDB data in 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; ``` **To export a DynamoDB table to an Amazon S3 bucket using formatting** + Create an external table that references a location in Amazon S3. This is shown below as s3\$1export. During the CREATE call, specify row formatting for the table. Then, when you use INSERT OVERWRITE to export data from DynamoDB to s3\$1export, the data is written out in the specified format. In the following example, the data is written out as comma-separated values (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; ``` **To export a DynamoDB table to an Amazon S3 bucket without specifying a column mapping** + Create a Hive table that references data stored in DynamoDB. This is similar to the preceding example, except that you are not specifying a column mapping. The table must have exactly one column of type `map`. If you then create an `EXTERNAL` table in Amazon S3 you can call the `INSERT OVERWRITE` command to write the data from DynamoDB to Amazon S3. You can use this to create an archive of your DynamoDB data in Amazon S3. Because there is no column mapping, you cannot query tables that are exported this way. Exporting data without specifying a column mapping is available in Hive 0.8.1.5 or later, which is supported on Amazon EMR AMI 2.2.*x* and later. ``` 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; ``` **To export a DynamoDB table to an Amazon S3 bucket using data compression** + Hive provides several compression codecs you can set during your Hive session. Doing so causes the exported data to be compressed in the specified format. The following example compresses the exported files using the Lempel-Ziv-Oberhumer (LZO) algorithm. ``` 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; ``` The available compression codecs are: + 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 **To export a DynamoDB table to HDFS** + Use the following Hive command, where *hdfs:///directoryName* is a valid HDFS path and *hiveTableName* is a table in Hive that references DynamoDB. This export operation is faster than exporting a DynamoDB table to Amazon S3 because Hive 0.7.1.1 uses HDFS as an intermediate step when exporting data to Amazon S3. The following example also shows how to set `dynamodb.throughput.read.percent` to 1.0 in order to increase the read request rate. ``` 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; ``` You can also export data to HDFS using formatting and compression as shown above for the export to Amazon S3. To do so, simply replace the Amazon S3 directory in the examples above with an HDFS directory. **To read non-printable UTF-8 character data in Hive** + You can read and write non-printable UTF-8 character data with Hive by using the `STORED AS SEQUENCEFILE` clause when you create the table. A SequenceFile is Hadoop binary file format; you need to use Hadoop to read this file. The following example shows how to export data from DynamoDB into Amazon S3. You can use this functionality to handle non-printable UTF-8 encoded characters. ``` 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; ``` ## Importing data to DynamoDB When you write data to DynamoDB using Hive you should ensure that the number of write capacity units is greater than the number of mappers in the cluster. For example, clusters that run on m1.xlarge EC2 instances produce 8 mappers per instance. In the case of a cluster that has 10 instances, that would mean a total of 80 mappers. If your write capacity units are not greater than the number of mappers in the cluster, the Hive write operation may consume all of the write throughput, or attempt to consume more throughput than is provisioned. For more information about the number of mappers produced by each EC2 instance type, see [Configure Hadoop](emr-hadoop-config.md). The number of mappers in Hadoop are controlled by the input splits. If there are too few splits, your write command might not be able to consume all the write throughput available. If an item with the same key exists in the target DynamoDB table, it is overwritten. If no item with the key exists in the target DynamoDB table, the item is inserted. **To import a table from Amazon S3 to DynamoDB** + You can use Amazon EMR (Amazon EMR) and Hive to write data from Amazon S3 to 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; ``` **To import a table from an Amazon S3 bucket to DynamoDB without specifying a column mapping** + Create an `EXTERNAL` table that references data stored in Amazon S3 that was previously exported from DynamoDB. Before importing, ensure that the table exists in DynamoDB and that it has the same key schema as the previously exported DynamoDB table. In addition, the table must have exactly one column of type `map`. If you then create a Hive table that is linked to DynamoDB, you can call the `INSERT OVERWRITE` command to write the data from Amazon S3 to DynamoDB. Because there is no column mapping, you cannot query tables that are imported this way. Importing data without specifying a column mapping is available in Hive 0.8.1.5 or later, which is supported on Amazon EMR AMI 2.2.3 and later. ``` 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; ``` **To import a table from HDFS to DynamoDB** + You can use Amazon EMR and Hive to write data from HDFS to 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; ``` ## Querying data in DynamoDB The following examples show the various ways you can use Amazon EMR to query data stored in DynamoDB. **To find the largest value for a mapped column (`max`)** + Use Hive commands like the following. In the first command, the CREATE statement creates a Hive table that references data stored in DynamoDB. The SELECT statement then uses that table to query data stored in DynamoDB. The following example finds the largest order placed by a given customer. ``` 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; ``` **To aggregate data using the `GROUP BY` clause** + You can use the `GROUP BY` clause to collect data across multiple records. This is often used with an aggregate function such as sum, count, min, or max. The following example returns a list of the largest orders from customers who have placed more than three orders. ``` 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; ``` **To join two DynamoDB tables** + The following example maps two Hive tables to data stored in DynamoDB. It then calls a join across those two tables. The join is computed on the cluster and returned. The join does not take place in DynamoDB. This example returns a list of customers and their purchases for customers that have placed more than two orders. ``` 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; ``` **To join two tables from different sources** + In the following example, Customer\$1S3 is a Hive table that loads a CSV file stored in Amazon S3 and hive\$1purchases is a table that references data in DynamoDB. The following example joins together customer data stored as a CSV file in Amazon S3 with order data stored in DynamoDB to return a set of data that represents orders placed by customers who have "Miller" in their name. ``` 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** In the preceding examples, the CREATE TABLE statements were included in each example for clarity and completeness. When running multiple queries or export operations against a given Hive table, you only need to create the table one time, at the beginning of the Hive session. # Optimizing performance for Amazon EMR operations in DynamoDB Amazon EMR operations on a DynamoDB table count as read operations, and are subject to the table's provisioned throughput settings. Amazon EMR implements its own logic to try to balance the load on your DynamoDB table to minimize the possibility of exceeding your provisioned throughput. At the end of each Hive query, Amazon EMR returns information about the cluster used to process the query, including how many times your provisioned throughput was exceeded. You can use this information, as well as CloudWatch metrics about your DynamoDB throughput, to better manage the load on your DynamoDB table in subsequent requests. The following factors influence Hive query performance when working with DynamoDB tables. ## Provisioned read capacity units When you run Hive queries against a DynamoDB table, you need to ensure that you have provisioned a sufficient amount of read capacity units. For example, suppose that you have provisioned 100 units of Read Capacity for your DynamoDB table. This will let you perform 100 reads, or 409,600 bytes, per second. If that table contains 20GB of data (21,474,836,480 bytes), and your Hive query performs a full table scan, you can estimate how long the query will take to run: * 21,474,836,480 / 409,600 = 52,429 seconds = 14.56 hours * The only way to decrease the time required would be to adjust the read capacity units on the source DynamoDB table. Adding more nodes to the Amazon EMR cluster will not help. In the Hive output, the completion percentage is updated when one or more mapper processes are finished. For a large DynamoDB table with a low provisioned Read Capacity setting, the completion percentage output might not be updated for a long time; in the case above, the job will appear to be 0% complete for several hours. For more detailed status on your job's progress, go to the Amazon EMR console; you will be able to view the individual mapper task status, and statistics for data reads. You can also log on to Hadoop interface on the master node and see the Hadoop statistics. This shows you the individual map task status and some data read statistics. For more information, see [Web interfaces hosted on the master node](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-web-interfaces.html) in the *Amazon EMR Management Guide*. ## Read percent setting By default, Amazon EMR manages the request load against your DynamoDB table according to your current provisioned throughput. However, when Amazon EMR returns information about your job that includes a high number of provisioned throughput exceeded responses, you can adjust the default read rate using the `dynamodb.throughput.read.percent` parameter when you set up the Hive table. For more information about setting the read percent parameter, see [Hive options](EMR_Interactive_Hive.md#EMR_Hive_Options). ## Write percent setting By default, Amazon EMR manages the request load against your DynamoDB table according to your current provisioned throughput. However, when Amazon EMR returns information about your job that includes a high number of provisioned throughput exceeded responses, you can adjust the default write rate using the `dynamodb.throughput.write.percent` parameter when you set up the Hive table. For more information about setting the write percent parameter, see [Hive options](EMR_Interactive_Hive.md#EMR_Hive_Options). ## Retry duration setting By default, Amazon EMR re-runs a Hive query if it has not returned a result within two minutes, the default retry interval. You can adjust this interval by setting the `dynamodb.retry.duration` parameter when you run a Hive query. For more information about setting the write percent parameter, see [Hive options](EMR_Interactive_Hive.md#EMR_Hive_Options). ## Number of map tasks The mapper daemons that Hadoop launches to process your requests to export and query data stored in DynamoDB are capped at a maximum read rate of 1 MiB per second to limit the read capacity used. If you have additional provisioned throughput available on DynamoDB, you can improve the performance of Hive export and query operations by increasing the number of mapper daemons. To do this, you can either increase the number of EC2 instances in your cluster *or* increase the number of mapper daemons running on each EC2 instance. You can increase the number of EC2 instances in a cluster by stopping the current cluster and re-launching it with a larger number of EC2 instances. You specify the number of EC2 instances in the **Configure EC2 Instances** dialog box if you're launching the cluster from the Amazon EMR console, or with the `--num-instances` option if you're launching the cluster from the CLI. The number of map tasks run on an instance depends on the EC2 instance type. For more information about the supported EC2 instance types and the number of mappers each one provides, see [Task configuration](emr-hadoop-task-config.md). There, you will find a "Task Configuration" section for each of the supported configurations. Another way to increase the number of mapper daemons is to change the `mapreduce.tasktracker.map.tasks.maximum` configuration parameter of Hadoop to a higher value. This has the advantage of giving you more mappers without increasing either the number or the size of EC2 instances, which saves you money. A disadvantage is that setting this value too high can cause the EC2 instances in your cluster to run out of memory. To set `mapreduce.tasktracker.map.tasks.maximum`, launch the cluster and specify a value for `mapreduce.tasktracker.map.tasks.maximum` as a property of the mapred-site configuration classification. This is shown in the following example. For more information, see [Configure applications](emr-configure-apps.md). ``` { "configurations": [ { "classification": "mapred-site", "properties": { "mapred.tasktracker.map.tasks.maximum": "10" } } ] } ``` ## Parallel data requests Multiple data requests, either from more than one user or more than one application to a single table may drain read provisioned throughput and slow performance. ## Process duration Data consistency in DynamoDB depends on the order of read and write operations on each node. While a Hive query is in progress, another application might load new data into the DynamoDB table or modify or delete existing data. In this case, the results of the Hive query might not reflect changes made to the data while the query was running. ## Avoid exceeding throughput When running Hive queries against DynamoDB, take care not to exceed your provisioned throughput, because this will deplete capacity needed for your application's calls to `DynamoDB::Get`. To ensure that this is not occurring, you should regularly monitor the read volume and throttling on application calls to `DynamoDB::Get` by checking logs and monitoring metrics in Amazon CloudWatch. ## Request time Scheduling Hive queries that access a DynamoDB table when there is lower demand on the DynamoDB table improves performance. For example, if most of your application's users live in San Francisco, you might choose to export daily data at 4 a.m. PST, when the majority of users are asleep, and not updating records in your DynamoDB database. ## Time-based tables If the data is organized as a series of time-based DynamoDB tables, such as one table per day, you can export the data when the table becomes no longer active. You can use this technique to back up data to Amazon S3 on an ongoing fashion. ## Archived data If you plan to run many Hive queries against the data stored in DynamoDB and your application can tolerate archived data, you may want to export the data to HDFS or Amazon S3 and run the Hive queries against a copy of the data instead of DynamoDB. This conserves your read operations and provisioned throughput.