}
```
The following sample highlights the sections used for condition expression.
![\[Get started with AWS DMS\]](http://docs.aws.amazon.com/dms/latest/userguide/images/datarep-Tasks-conditional1.png)
### Using attribute mapping with object mapping
Attribute mapping lets you specify a template string using source column names to restructure data on the target. There is no formatting done other than what the user specifies in the template.
The following example shows the structure of the source database and the desired structure of the DynamoDB target. First is shown the structure of the source, in this case an Oracle database, and then the desired structure of the data in DynamoDB. The example concludes with the JSON used to create the desired target structure.
The structure of the Oracle data is as follows:
****
| FirstName | LastName | StoreId | HomeAddress | HomePhone | WorkAddress | WorkPhone | DateOfBirth |
| --- | --- | --- | --- | --- | --- | --- | --- |
| Primary Key | N/A | |
| Randy | Marsh | 5 | 221B Baker Street | 1234567890 | 31 Spooner Street, Quahog | 9876543210 | 02/29/1988 |
The structure of the DynamoDB data is as follows:
****
| CustomerName | StoreId | ContactDetails | DateOfBirth |
| --- | --- | --- | --- |
| Partition Key | Sort Key | N/A |
| Randy,Marsh
| 5
| {
"Name": "Randy",
"Home": {
"Address": "221B Baker Street",
"Phone": 1234567890
},
"Work": {
"Address": "31 Spooner Street, Quahog",
"Phone": 9876541230
}
} | 02/29/1988
|
The following JSON shows the object mapping and column mapping used to achieve the DynamoDB structure:
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "test",
"table-name": "%"
},
"rule-action": "include"
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "TransformToDDB",
"rule-action": "map-record-to-record",
"object-locator": {
"schema-name": "test",
"table-name": "customer"
},
"target-table-name": "customer_t",
"mapping-parameters": {
"partition-key-name": "CustomerName",
"sort-key-name": "StoreId",
"exclude-columns": [
"FirstName",
"LastName",
"HomeAddress",
"HomePhone",
"WorkAddress",
"WorkPhone"
],
"attribute-mappings": [
{
"target-attribute-name": "CustomerName",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${FirstName},${LastName}"
},
{
"target-attribute-name": "StoreId",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${StoreId}"
},
{
"target-attribute-name": "ContactDetails",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "{\"Name\":\"${FirstName}\",\"Home\":{\"Address\":\"${HomeAddress}\",\"Phone\":\"${HomePhone}\"}, \"Work\":{\"Address\":\"${WorkAddress}\",\"Phone\":\"${WorkPhone}\"}}"
}
]
}
}
]
}
```
Another way to use column mapping is to use DynamoDB format as your document type. The following code example uses *dynamodb-map* as the `attribute-sub-type` for attribute mapping.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "test",
"table-name": "%"
},
"rule-action": "include"
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "TransformToDDB",
"rule-action": "map-record-to-record",
"object-locator": {
"schema-name": "test",
"table-name": "customer"
},
"target-table-name": "customer_t",
"mapping-parameters": {
"partition-key-name": "CustomerName",
"sort-key-name": "StoreId",
"exclude-columns": [
"FirstName",
"LastName",
"HomeAddress",
"HomePhone",
"WorkAddress",
"WorkPhone"
],
"attribute-mappings": [
{
"target-attribute-name": "CustomerName",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${FirstName},${LastName}"
},
{
"target-attribute-name": "StoreId",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${StoreId}"
},
{
"target-attribute-name": "ContactDetails",
"attribute-type": "document",
"attribute-sub-type": "dynamodb-map",
"value": {
"M": {
"Name": {
"S": "${FirstName}"
},
"Home": {
"M": {
"Address": {
"S": "${HomeAddress}"
},
"Phone": {
"S": "${HomePhone}"
}
}
},
"Work": {
"M": {
"Address": {
"S": "${WorkAddress}"
},
"Phone": {
"S": "${WorkPhone}"
}
}
}
}
}
}
]
}
}
]
}
```
As an alternative to `dynamodb-map`, you can use `dynamodb-list` as the attribute-sub-type for attribute mapping, as shown in the following example.
```
{
"target-attribute-name": "ContactDetailsList",
"attribute-type": "document",
"attribute-sub-type": "dynamodb-list",
"value": {
"L": [
{
"N": "${FirstName}"
},
{
"N": "${HomeAddress}"
},
{
"N": "${HomePhone}"
},
{
"N": "${WorkAddress}"
},
{
"N": "${WorkPhone}"
}
]
}
}
```
### Example 1: Using attribute mapping with object mapping
The following example migrates data from two MySQL database tables, *nfl\$1data* and *sport\$1team* , to two DynamoDB table called *NFLTeams* and *SportTeams*. The structure of the tables and the JSON used to map the data from the MySQL database tables to the DynamoDB tables are shown following.
The structure of the MySQL database table *nfl\$1data* is shown below:
```
mysql> desc nfl_data;
+---------------+-------------+------+-----+---------+-------+
| Field | Type | Null | Key | Default | Extra |
+---------------+-------------+------+-----+---------+-------+
| Position | varchar(5) | YES | | NULL | |
| player_number | smallint(6) | YES | | NULL | |
| Name | varchar(40) | YES | | NULL | |
| status | varchar(10) | YES | | NULL | |
| stat1 | varchar(10) | YES | | NULL | |
| stat1_val | varchar(10) | YES | | NULL | |
| stat2 | varchar(10) | YES | | NULL | |
| stat2_val | varchar(10) | YES | | NULL | |
| stat3 | varchar(10) | YES | | NULL | |
| stat3_val | varchar(10) | YES | | NULL | |
| stat4 | varchar(10) | YES | | NULL | |
| stat4_val | varchar(10) | YES | | NULL | |
| team | varchar(10) | YES | | NULL | |
+---------------+-------------+------+-----+---------+-------+
```
The structure of the MySQL database table *sport\$1team* is shown below:
```
mysql> desc sport_team;
+---------------------------+--------------+------+-----+---------+----------------+
| Field | Type | Null | Key | Default | Extra |
+---------------------------+--------------+------+-----+---------+----------------+
| id | mediumint(9) | NO | PRI | NULL | auto_increment |
| name | varchar(30) | NO | | NULL | |
| abbreviated_name | varchar(10) | YES | | NULL | |
| home_field_id | smallint(6) | YES | MUL | NULL | |
| sport_type_name | varchar(15) | NO | MUL | NULL | |
| sport_league_short_name | varchar(10) | NO | | NULL | |
| sport_division_short_name | varchar(10) | YES | | NULL | |
```
The table-mapping rules used to map the two tables to the two DynamoDB tables is shown below:
```
{
"rules":[
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "dms_sample",
"table-name": "nfl_data"
},
"rule-action": "include"
},
{
"rule-type": "selection",
"rule-id": "2",
"rule-name": "2",
"object-locator": {
"schema-name": "dms_sample",
"table-name": "sport_team"
},
"rule-action": "include"
},
{
"rule-type":"object-mapping",
"rule-id":"3",
"rule-name":"MapNFLData",
"rule-action":"map-record-to-record",
"object-locator":{
"schema-name":"dms_sample",
"table-name":"nfl_data"
},
"target-table-name":"NFLTeams",
"mapping-parameters":{
"partition-key-name":"Team",
"sort-key-name":"PlayerName",
"exclude-columns": [
"player_number", "team", "name"
],
"attribute-mappings":[
{
"target-attribute-name":"Team",
"attribute-type":"scalar",
"attribute-sub-type":"string",
"value":"${team}"
},
{
"target-attribute-name":"PlayerName",
"attribute-type":"scalar",
"attribute-sub-type":"string",
"value":"${name}"
},
{
"target-attribute-name":"PlayerInfo",
"attribute-type":"scalar",
"attribute-sub-type":"string",
"value":"{\"Number\": \"${player_number}\",\"Position\": \"${Position}\",\"Status\": \"${status}\",\"Stats\": {\"Stat1\": \"${stat1}:${stat1_val}\",\"Stat2\": \"${stat2}:${stat2_val}\",\"Stat3\": \"${stat3}:${
stat3_val}\",\"Stat4\": \"${stat4}:${stat4_val}\"}"
}
]
}
},
{
"rule-type":"object-mapping",
"rule-id":"4",
"rule-name":"MapSportTeam",
"rule-action":"map-record-to-record",
"object-locator":{
"schema-name":"dms_sample",
"table-name":"sport_team"
},
"target-table-name":"SportTeams",
"mapping-parameters":{
"partition-key-name":"TeamName",
"exclude-columns": [
"name", "id"
],
"attribute-mappings":[
{
"target-attribute-name":"TeamName",
"attribute-type":"scalar",
"attribute-sub-type":"string",
"value":"${name}"
},
{
"target-attribute-name":"TeamInfo",
"attribute-type":"scalar",
"attribute-sub-type":"string",
"value":"{\"League\": \"${sport_league_short_name}\",\"Division\": \"${sport_division_short_name}\"}"
}
]
}
}
]
}
```
The sample output for the *NFLTeams* DynamoDB table is shown below:
```
"PlayerInfo": "{\"Number\": \"6\",\"Position\": \"P\",\"Status\": \"ACT\",\"Stats\": {\"Stat1\": \"PUNTS:73\",\"Stat2\": \"AVG:46\",\"Stat3\": \"LNG:67\",\"Stat4\": \"IN 20:31\"}",
"PlayerName": "Allen, Ryan",
"Position": "P",
"stat1": "PUNTS",
"stat1_val": "73",
"stat2": "AVG",
"stat2_val": "46",
"stat3": "LNG",
"stat3_val": "67",
"stat4": "IN 20",
"stat4_val": "31",
"status": "ACT",
"Team": "NE"
}
```
The sample output for the SportsTeams *DynamoDB* table is shown below:
```
{
"abbreviated_name": "IND",
"home_field_id": 53,
"sport_division_short_name": "AFC South",
"sport_league_short_name": "NFL",
"sport_type_name": "football",
"TeamInfo": "{\"League\": \"NFL\",\"Division\": \"AFC South\"}",
"TeamName": "Indianapolis Colts"
}
```
## Target data types for DynamoDB
The DynamoDB endpoint for AWS DMS supports most DynamoDB data types. The following table shows the Amazon AWS DMS target data types that are supported when using AWS DMS and the default mapping from AWS DMS data types.
For additional information about AWS DMS data types, see [Data types for AWS Database Migration Service](CHAP_Reference.DataTypes.md).
When AWS DMS migrates data from heterogeneous databases, we map data types from the source database to intermediate data types called AWS DMS data types. We then map the intermediate data types to the target data types. The following table shows each AWS DMS data type and the data type it maps to in DynamoDB:
| AWS DMS data type | DynamoDB data type |
| --- | --- |
| String | String |
| WString | String |
| Boolean | Boolean |
| Date | String |
| DateTime | String |
| INT1 | Number |
| INT2 | Number |
| INT4 | Number |
| INT8 | Number |
| Numeric | Number |
| Real4 | Number |
| Real8 | Number |
| UINT1 | Number |
| UINT2 | Number |
| UINT4 | Number |
| UINT8 | Number |
| CLOB | String |
# Using Amazon Kinesis Data Streams as a target for AWS Database Migration Service
You can use AWS DMS to migrate data to an Amazon Kinesis data stream. Amazon Kinesis data streams are part of the Amazon Kinesis Data Streams service. You can use Kinesis data streams to collect and process large streams of data records in real time.
A Kinesis data stream is made up of shards. *Shards* are uniquely identified sequences of data records in a stream. For more information on shards in Amazon Kinesis Data Streams, see [Shard](https://docs.aws.amazon.com/streams/latest/dev/key-concepts.html#shard) in the *Amazon Kinesis Data Streams Developer Guide.*
AWS Database Migration Service publishes records to a Kinesis data stream using JSON. During conversion, AWS DMS serializes each record from the source database into an attribute-value pair in JSON format or a JSON\$1UNFORMATTED message format. A JSON\$1UNFORMATTED message format is a single line JSON string with new line delimiter. It allows Amazon Data Firehose to deliver Kinesis data to an Amazon S3 destination, and then query it using various query engines including Amazon Athena.
You use object mapping to migrate your data from any supported data source to a target stream. With object mapping, you determine how to structure the data records in the stream. You also define a partition key for each table, which Kinesis Data Streams uses to group the data into its shards.
AWS DMS also sets several Kinesis Data Streams parameter values. The cost for the table creation depends on the amount of data and the number of tables to be migrated.
**Note**
The **SSL Mode** option on the AWS DMS console or API doesn’t apply to some data streaming and NoSQL services like Kinesis and DynamoDB. They are secure by default, so AWS DMS shows the SSL mode setting is equal to none (**SSL Mode=None**). You don’t need to provide any additional configuration for your endpoint to make use of SSL. For example, when using Kinesis as a target endpoint, it is secure by default. All API calls to Kinesis use SSL, so there is no need for an additional SSL option in the AWS DMS endpoint. You can securely put data and retrieve data through SSL endpoints using the HTTPS protocol, which AWS DMS uses by default when connecting to a Kinesis Data Stream.
**Kinesis Data Streams endpoint settings**
When you use Kinesis Data Streams target endpoints, you can get transaction and control details using the `KinesisSettings` option in the AWS DMS API.
You can set connection settings in the following ways:
+ In the AWS DMS console, using endpoint settings.
+ In the CLI, using the `kinesis-settings` option of the [CreateEndpoint](https://docs.aws.amazon.com/dms/latest/APIReference/API_CreateEndpoint.html) command.
In the CLI, use the following request parameters of the `kinesis-settings` option:
**Note**
Support for the `IncludeNullAndEmpty` endpoint setting is available in AWS DMS version 3.4.1 and higher. But support for the other following endpoint settings for Kinesis Data Streams targets is available in AWS DMS.
+ `MessageFormat` – The output format for the records created on the endpoint. The message format is `JSON` (default) or `JSON_UNFORMATTED` (a single line with no tab).
+ `IncludeControlDetails` – Shows detailed control information for table definition, column definition, and table and column changes in the Kinesis message output. The default is `false`.
+ `IncludeNullAndEmpty` – Include NULL and empty columns in the target. The default is `false`.
+ `IncludePartitionValue` – Shows the partition value within the Kinesis message output, unless the partition type is `schema-table-type`. The default is `false`.
+ `IncludeTableAlterOperations` – Includes any data definition language (DDL) operations that change the table in the control data, such as `rename-table`, `drop-table`, `add-column`, `drop-column`, and `rename-column`. The default is `false`.
+ `IncludeTransactionDetails` – Provides detailed transaction information from the source database. This information includes a commit timestamp, a log position, and values for `transaction_id`, `previous_transaction_id`, and `transaction_record_id `(the record offset within a transaction). The default is `false`.
+ `PartitionIncludeSchemaTable` – Prefixes schema and table names to partition values, when the partition type is `primary-key-type`. Doing this increases data distribution among Kinesis shards. For example, suppose that a `SysBench` schema has thousands of tables and each table has only limited range for a primary key. In this case, the same primary key is sent from thousands of tables to the same shard, which causes throttling. The default is `false`.
+ `UseLargeIntegerValue` – Use up to 18 digit int instead of casting ints as doubles, available from AWS DMS version 3.5.4. The default is false.
The following example shows the `kinesis-settings` option in use with an example `create-endpoint` command issued using the AWS CLI.
```
aws dms \
create-endpoint \
--region \
--endpoint-identifier \
--endpoint-type target \
--engine-name kinesis \
--kinesis-settings ServiceAccessRoleArn=arn:aws:iam:::role/,StreamArn=arn:aws:kinesis:::stream/,MessageFormat=json-unformatted,
IncludeControlDetails=true,IncludeTransactionDetails=true,IncludePartitionValue=true,PartitionIncludeSchemaTable=true,
IncludeTableAlterOperations=true
```
**Multithreaded full load task settings**
To help increase the speed of the transfer, AWS DMS supports a multithreaded full load to a Kinesis Data Streams target instance. DMS supports this multithreading with task settings that include the following:
+ `MaxFullLoadSubTasks` – Use this option to indicate the maximum number of source tables to load in parallel. DMS loads each table into its corresponding Kinesis target table using a dedicated subtask. The default is 8; the maximum value is 49.
+ `ParallelLoadThreads` – Use this option to specify the number of threads that AWS DMS uses to load each table into its Kinesis target table. The maximum value for a Kinesis Data Streams target is 32. You can ask to have this maximum limit increased.
+ `ParallelLoadBufferSize` – Use this option to specify the maximum number of records to store in the buffer that the parallel load threads use to load data to the Kinesis target. The default value is 50. The maximum value is 1,000. Use this setting with `ParallelLoadThreads`. `ParallelLoadBufferSize` is valid only when there is more than one thread.
+ `ParallelLoadQueuesPerThread` – Use this option to specify the number of queues each concurrent thread accesses to take data records out of queues and generate a batch load for the target. The default is 1. However, for Kinesis targets of various payload sizes, the valid range is 5–512 queues per thread.
**Multithreaded CDC load task settings**
You can improve the performance of change data capture (CDC) for real-time data streaming target endpoints like Kinesis using task settings to modify the behavior of the `PutRecords` API call. To do this, you can specify the number of concurrent threads, queues per thread, and the number of records to store in a buffer using `ParallelApply*` task settings. For example, suppose you want to perform a CDC load and apply 128 threads in parallel. You also want to access 64 queues per thread, with 50 records stored per buffer.
To promote CDC performance, AWS DMS supports these task settings:
+ `ParallelApplyThreads` – Specifies the number of concurrent threads that AWS DMS uses during a CDC load to push data records to a Kinesis target endpoint. The default value is zero (0) and the maximum value is 32.
+ `ParallelApplyBufferSize` – Specifies the maximum number of records to store in each buffer queue for concurrent threads to push to a Kinesis target endpoint during a CDC load. The default value is 100 and the maximum value is 1,000. Use this option when `ParallelApplyThreads` specifies more than one thread.
+ `ParallelApplyQueuesPerThread` – Specifies the number of queues that each thread accesses to take data records out of queues and generate a batch load for a Kinesis endpoint during CDC. The default value is 1 and the maximum value is 512.
When using `ParallelApply*` task settings, the `partition-key-type` default is the `primary-key` of the table, not `schema-name.table-name`.
## Using a before image to view original values of CDC rows for a Kinesis data stream as a target
When writing CDC updates to a data-streaming target like Kinesis, you can view a source database row's original values before change by an update. To make this possible, AWS DMS populates a *before image* of update events based on data supplied by the source database engine.
Different source database engines provide different amounts of information for a before image:
+ Oracle provides updates to columns only if they change.
+ PostgreSQL provides only data for columns that are part of the primary key (changed or not). To provide data for all columns (changed or not), you need to set `REPLICA_IDENTITY` to `FULL` instead of `DEFAULT`. Note that you should choose the `REPLICA_IDENTITY` setting carefully for each table. If you set `REPLICA_IDENTITY` to `FULL`, all of the column values are written to write-ahead logging (WAL) continuously. This may cause performance or resource issues with tables that are updated frequently.
+ MySQL generally provides data for all columns except for BLOB and CLOB data types (changed or not).
To enable before imaging to add original values from the source database to the AWS DMS output, use either the `BeforeImageSettings` task setting or the `add-before-image-columns` parameter. This parameter applies a column transformation rule.
`BeforeImageSettings` adds a new JSON attribute to every update operation with values collected from the source database system, as shown following.
```
"BeforeImageSettings": {
"EnableBeforeImage": boolean,
"FieldName": string,
"ColumnFilter": pk-only (default) / non-lob / all (but only one)
}
```
**Note**
Only apply `BeforeImageSettings` to AWS DMS tasks that contain a CDC component, such as full load plus CDC tasks (which migrate existing data and replicate ongoing changes), or to CDC only tasks (which replicate data changes only). Don't apply `BeforeImageSettings` to tasks that are full load only.
For `BeforeImageSettings` options, the following applies:
+ Set the `EnableBeforeImage` option to `true` to enable before imaging. The default is `false`.
+ Use the `FieldName` option to assign a name to the new JSON attribute. When `EnableBeforeImage` is `true`, `FieldName` is required and can't be empty.
+ The `ColumnFilter` option specifies a column to add by using before imaging. To add only columns that are part of the table's primary keys, use the default value, `pk-only`. To add any column that has a before image value, use `all`. Note that the before image does not contain columns with LOB data types, such as CLOB or BLOB.
```
"BeforeImageSettings": {
"EnableBeforeImage": true,
"FieldName": "before-image",
"ColumnFilter": "pk-only"
}
```
**Note**
Amazon S3 targets don't support `BeforeImageSettings`. For S3 targets, use only the `add-before-image-columns` transformation rule to perform before imaging during CDC.
### Using a before image transformation rule
As as an alternative to task settings, you can use the `add-before-image-columns` parameter, which applies a column transformation rule. With this parameter, you can enable before imaging during CDC on data streaming targets like Kinesis.
By using `add-before-image-columns` in a transformation rule, you can apply more fine-grained control of the before image results. Transformation rules enable you to use an object locator that gives you control over tables selected for the rule. Also, you can chain transformation rules together, which allows different rules to be applied to different tables. You can then manipulate the columns produced by using other rules.
**Note**
Don't use the `add-before-image-columns` parameter together with the `BeforeImageSettings` task setting within the same task. Instead, use either the parameter or the setting, but not both, for a single task.
A `transformation` rule type with the `add-before-image-columns` parameter for a column must provide a `before-image-def` section. The following shows an example.
```
{
"rule-type": "transformation",
…
"rule-target": "column",
"rule-action": "add-before-image-columns",
"before-image-def":{
"column-filter": one-of (pk-only / non-lob / all),
"column-prefix": string,
"column-suffix": string,
}
}
```
The value of `column-prefix` is prepended to a column name, and the default value of `column-prefix` is `BI_`. The value of `column-suffix` is appended to the column name, and the default is empty. Don't set both `column-prefix` and `column-suffix` to empty strings.
Choose one value for `column-filter`. To add only columns that are part of table primary keys, choose `pk-only` . Choose `non-lob` to only add columns that are not of LOB type. Or choose `all` to add any column that has a before-image value.
### Example for a before image transformation rule
The transformation rule in the following example adds a new column called `BI_emp_no` in the target. So a statement like `UPDATE employees SET emp_no = 3 WHERE emp_no = 1;` populates the `BI_emp_no` field with 1. When you write CDC updates to Amazon S3 targets, the `BI_emp_no` column makes it possible to tell which original row was updated.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "%",
"table-name": "%"
},
"rule-action": "include"
},
{
"rule-type": "transformation",
"rule-id": "2",
"rule-name": "2",
"rule-target": "column",
"object-locator": {
"schema-name": "%",
"table-name": "employees"
},
"rule-action": "add-before-image-columns",
"before-image-def": {
"column-prefix": "BI_",
"column-suffix": "",
"column-filter": "pk-only"
}
}
]
}
```
For information on using the `add-before-image-columns` rule action, see [Transformation rules and actions](CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Transformations.md).
## Prerequisites for using a Kinesis data stream as a target for AWS Database Migration Service
### IAM role for using a Kinesis data stream as a target for AWS Database Migration Service
Before you set up a Kinesis data stream as a target for AWS DMS, make sure that you create an IAM role. This role must allow AWS DMS to assume and grant access to the Kinesis data streams that are being migrated into. The minimum set of access permissions is shown in the following IAM policy.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "1",
"Effect": "Allow",
"Principal": {
"Service": "dms.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
The role that you use for the migration to a Kinesis data stream must have the following permissions.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kinesis:DescribeStream",
"kinesis:PutRecord",
"kinesis:PutRecords"
],
"Resource": "*"
}
]
}
```
------
### Accessing a Kinesis data stream as a target for AWS Database Migration Service
In AWS DMS version 3.4.7 and higher, to connect to an Kinesis endpoint, you must do one of the following:
+ Configure DMS to use VPC endpoints. For information about configuring DMS to use VPC endpoints, see [Configuring VPC endpoints for AWS DMS](CHAP_VPC_Endpoints.md).
+ Configure DMS to use public routes, that is, make your replication instance public. For information about public replication instances, see [Public and private replication instances](CHAP_ReplicationInstance.PublicPrivate.md).
## Limitations when using Kinesis Data Streams as a target for AWS Database Migration Service
The following limitations apply when using Kinesis Data Streams as a target:
+ AWS DMS publishes each update to a single record in the source database as one data record in a given Kinesis data stream regardless of transactions. However, you can include transaction details for each data record by using relevant parameters of the `KinesisSettings` API.
+ Full LOB mode is not supported.
+ The maximum supported LOB size is 1 MB.
+ Kinesis Data Streams don't support deduplication. Applications that consume data from a stream need to handle duplicate records. For more information, see [Handling duplicate records](https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-duplicates.html) in the *Amazon Kinesis Data Streams Developer Guide.*
+ AWS DMS supports the following four forms for partition keys:
+ `SchemaName.TableName`: A combination of the schema and table name.
+ `${AttributeName}`: The value of one of the fields in the JSON, or the primary key of the table in the source database.
+ `transaction-id`: The CDC transaction ID. All records within the same transaction go to the same partition.
+ `constant`: A fixed literal value for every record regardless of table or data. All records are sent to the same partition key value "constant", providing strict global ordering across all tables.
```
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "PartitionKeyTypeExample",
"rule-action": "map-record-to-document",
"object-locator": {
"schema-name": "onprem",
"table-name": "it_system"
},
"mapping-parameters": {
"partition-key-type": "transaction-id | constant | attribute-name | schema-table"
}
}
```
+ For information about encrypting your data at rest within Kinesis Data Streams, see [Data protection in Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/server-side-encryption.html.html) in the *AWS Key Management Service Developer Guide*.
+ The `IncludeTransactionDetails` endpoint setting is only supported when the source endpoint is Oracle, SQL Server, PostgreSQL, or MySQL. For other source endpoint types, transaction details will not be included.
+ `BatchApply` is not supported for a Kinesis endpoint. Using Batch Apply (for example, the `BatchApplyEnabled` target metadata task setting) for a Kinesis target causes task failure and data loss. Do not enable `BatchApply` when using Kinesis as a target endpoint.
+ Kinesis targets are only supported for a Kinesis data stream in the same AWS account and the same AWS Region as the replication instance.
+ When migrating from a MySQL source, the BeforeImage data doesn't include CLOB and BLOB data types. For more information, see [Using a before image to view original values of CDC rows for a Kinesis data stream as a target](#CHAP_Target.Kinesis.BeforeImage).
+ AWS DMS doesn't support migrating values of `BigInt` data type with more than 16 digits. To work around this limitation, you can use the following transformation rule to convert the `BigInt` column to a string. For more information about transformation rules, see [Transformation rules and actions](CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Transformations.md).
```
{
"rule-type": "transformation",
"rule-id": "id",
"rule-name": "name",
"rule-target": "column",
"object-locator": {
"schema-name": "valid object-mapping rule action",
"table-name": "",
"column-name": ""
},
"rule-action": "change-data-type",
"data-type": {
"type": "string",
"length": 20
}
}
```
+ When multiple DML operations within a single transaction modify a Large Object (LOB) column on the source database, the target database retains only the final LOB value from the last operation in that transaction. The intermediate LOB values set by earlier operations in the same transaction are overwritten, which can result in potential data loss or inconsistencies. This behavior occurs due to how LOB data is processed during replication.
+ AWS DMS does not support source data containing embedded `'\0'` characters when using Kinesis as a target endpoint. Data containing embedded `'\0'` characters will be truncated at the first `'\0'` character.
## Using object mapping to migrate data to a Kinesis data stream
AWS DMS uses table-mapping rules to map data from the source to the target Kinesis data stream. To map data to a target stream, you use a type of table-mapping rule called object mapping. You use object mapping to define how data records in the source map to the data records published to the Kinesis data stream.
Kinesis data streams don't have a preset structure other than having a partition key. In an object mapping rule, the possible values of a `partition-key-type` for data records are `schema-table`, `transaction-id`, `primary-key`, `constant`, and `attribute-name`.
To create an object-mapping rule, you specify `rule-type` as `object-mapping`. This rule specifies what type of object mapping you want to use.
The structure for the rule is as follows.
```
{
"rules": [
{
"rule-type": "object-mapping",
"rule-id": "id",
"rule-name": "name",
"rule-action": "valid object-mapping rule action",
"object-locator": {
"schema-name": "case-sensitive schema name",
"table-name": ""
}
}
]
}
```
AWS DMS currently supports `map-record-to-record` and `map-record-to-document` as the only valid values for the `rule-action` parameter. These settings affect values that aren't excluded as part of the `exclude-columns` attribute list. The `map-record-to-record` and `map-record-to-document` values specify how AWS DMS handles these records by default. These values don't affect the attribute mappings in any way.
Use `map-record-to-record` when migrating from a relational database to a Kinesis data stream. This rule type uses the `taskResourceId.schemaName.tableName` value from the relational database as the partition key in the Kinesis data stream and creates an attribute for each column in the source database.
When using `map-record-to-record`, note the following:
+ This setting only affects columns excluded by the `exclude-columns` list.
+ For every such column, AWS DMS creates a corresponding attribute in the target topic.
+ AWS DMS creates this corresponding attribute regardless of whether the source column is used in an attribute mapping.
Use `map-record-to-document` to put source columns into a single, flat document in the appropriate target stream using the attribute name "\$1doc". AWS DMS places the data into a single, flat map on the source called "`_doc`". This placement applies to any column in the source table not listed in the `exclude-columns` attribute list.
One way to understand `map-record-to-record` is to see it in action. For this example, assume that you are starting with a relational database table row with the following structure and data.
| FirstName | LastName | StoreId | HomeAddress | HomePhone | WorkAddress | WorkPhone | DateofBirth |
| --- | --- | --- | --- | --- | --- | --- | --- |
| Randy | Marsh | 5 | 221B Baker Street | 1234567890 | 31 Spooner Street, Quahog | 9876543210 | 02/29/1988 |
To migrate this information from a schema named `Test` to a Kinesis data stream, you create rules to map the data to the target stream. The following rule illustrates the mapping.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"rule-action": "include",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
}
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "DefaultMapToKinesis",
"rule-action": "map-record-to-record",
"object-locator": {
"schema-name": "Test",
"table-name": "Customers"
}
}
]
}
```
The following illustrates the resulting record format in the Kinesis data stream:
+ StreamName: XXX
+ PartitionKey: Test.Customers //schmaName.tableName
+ Data: //The following JSON message
```
{
"FirstName": "Randy",
"LastName": "Marsh",
"StoreId": "5",
"HomeAddress": "221B Baker Street",
"HomePhone": "1234567890",
"WorkAddress": "31 Spooner Street, Quahog",
"WorkPhone": "9876543210",
"DateOfBirth": "02/29/1988"
}
```
However, suppose that you use the same rules but change the `rule-action` parameter to `map-record-to-document` and exclude certain columns. The following rule illustrates the mapping.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"rule-action": "include",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
}
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "DefaultMapToKinesis",
"rule-action": "map-record-to-document",
"object-locator": {
"schema-name": "Test",
"table-name": "Customers"
},
"mapping-parameters": {
"exclude-columns": [
"homeaddress",
"homephone",
"workaddress",
"workphone"
]
}
}
]
}
```
In this case, the columns not listed in the `exclude-columns` parameter, `FirstName`, `LastName`, `StoreId` and `DateOfBirth`, are mapped to `_doc`. The following illustrates the resulting record format.
```
{
"data":{
"_doc":{
"FirstName": "Randy",
"LastName": "Marsh",
"StoreId": "5",
"DateOfBirth": "02/29/1988"
}
}
}
```
### Restructuring data with attribute mapping
You can restructure the data while you are migrating it to a Kinesis data stream using an attribute map. For example, you might want to combine several fields in the source into a single field in the target. The following attribute map illustrates how to restructure the data.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"rule-action": "include",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
}
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "TransformToKinesis",
"rule-action": "map-record-to-record",
"target-table-name": "CustomerData",
"object-locator": {
"schema-name": "Test",
"table-name": "Customers"
},
"mapping-parameters": {
"partition-key-type": "attribute-name",
"partition-key-name": "CustomerName",
"exclude-columns": [
"firstname",
"lastname",
"homeaddress",
"homephone",
"workaddress",
"workphone"
],
"attribute-mappings": [
{
"target-attribute-name": "CustomerName",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${lastname}, ${firstname}"
},
{
"target-attribute-name": "ContactDetails",
"attribute-type": "document",
"attribute-sub-type": "json",
"value": {
"Home": {
"Address": "${homeaddress}",
"Phone": "${homephone}"
},
"Work": {
"Address": "${workaddress}",
"Phone": "${workphone}"
}
}
}
]
}
}
]
}
```
To set a constant value for `partition-key`, specify `"partition-key-type: "constant"`, this sets the partition value to `constant`. For example, you might do this to force all the data to be stored in a single shard. The following mapping illustrates this approach.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
},
"rule-action": "include"
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "TransformToKinesis",
"rule-action": "map-record-to-document",
"object-locator": {
"schema-name": "Test",
"table-name": "Customer"
},
"mapping-parameters": {
"partition-key-type": "constant",
"exclude-columns": [
"FirstName",
"LastName",
"HomeAddress",
"HomePhone",
"WorkAddress",
"WorkPhone"
],
"attribute-mappings": [
{
"target-attribute-name": "CustomerName",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${FirstName},${LastName}"
},
{
"target-attribute-name": "ContactDetails",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": {
"Home": {
"Address": "${HomeAddress}",
"Phone": "${HomePhone}"
},
"Work": {
"Address": "${WorkAddress}",
"Phone": "${WorkPhone}"
}
}
},
{
"target-attribute-name": "DateOfBirth",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${DateOfBirth}"
}
]
}
}
]
}
```
**Note**
The `partition-key` value for a control record that is for a specific table is `TaskId.SchemaName.TableName`. The `partition-key` value for a control record that is for a specific task is that record's `TaskId`. Specifying a `partition-key` value in the object mapping has no impact on the `partition-key` for a control record.
When `partition-key-type` is set to `attribute-name` in a table mapping rule, you must specify `partition-key-name`, which must reference either a column from the source table or a custom column defined in the mapping. Additionally, `attribute-mappings` must be provided to define how source columns map to the target Kinesis Stream.
### Message format for Kinesis Data Streams
The JSON output is simply a list of key-value pairs. A JSON\$1UNFORMATTED message format is a single line JSON string with new line delimiter.
AWS DMS provides the following reserved fields to make it easier to consume the data from the Kinesis Data Streams:
**RecordType**
The record type can be either data or control. *Data records *represent the actual rows in the source. *Control records* are for important events in the stream, for example a restart of the task.
**Operation**
For data records, the operation can be `load`, `insert`, `update`, or `delete`.
For control records, the operation can be `create-table`, `rename-table`, `drop-table`, `change-columns`, `add-column`, `drop-column`, `rename-column`, or `column-type-change`.
**SchemaName**
The source schema for the record. This field can be empty for a control record.
**TableName**
The source table for the record. This field can be empty for a control record.
**Timestamp**
The timestamp for when the JSON message was constructed. The field is formatted with the ISO 8601 format.
# Using Apache Kafka as a target for AWS Database Migration Service
You can use AWS DMS to migrate data to an Apache Kafka cluster. Apache Kafka is a distributed streaming platform. You can use Apache Kafka for ingesting and processing streaming data in real-time.
AWS also offers Amazon Managed Streaming for Apache Kafka (Amazon MSK) to use as an AWS DMS target. Amazon MSK is a fully managed Apache Kafka streaming service that simplifies the implementation and management of Apache Kafka instances. It works with open-source Apache Kafka versions, and you access Amazon MSK instances as AWS DMS targets exactly like any Apache Kafka instance. For more information, see [What is Amazon MSK?](https://docs.aws.amazon.com/msk/latest/developerguide/what-is-msk.html) in the* Amazon Managed Streaming for Apache Kafka Developer Guide.*
A Kafka cluster stores streams of records in categories called topics that are divided into partitions. *Partitions* are uniquely identified sequences of data records (messages) in a topic. Partitions can be distributed across multiple brokers in a cluster to enable parallel processing of a topic’s records. For more information on topics and partitions and their distribution in Apache Kafka, see [Topics and logs](https://kafka.apache.org/documentation/#intro_topics) and [Distribution](https://kafka.apache.org/documentation/#intro_distribution).
Your Kafka cluster can be either an Amazon MSK instance, a cluster running on an Amazon EC2 instance, or an on-premises cluster. An Amazon MSK instance or a cluster on an Amazon EC2 instance can be in the same VPC or a different one. If your cluster is on-premises, you can use your own on-premises name server for your replication instance to resolve the cluster's host name. For information about setting up a name server for your replication instance, see [Using your own on-premises name server](CHAP_BestPractices.md#CHAP_BestPractices.Rte53DNSResolver). For more information about setting up a network, see [Setting up a network for a replication instance](CHAP_ReplicationInstance.VPC.md).
When using an Amazon MSK cluster, make sure that its security group allows access from your replication instance. For information about changing the security group for an Amazon MSK cluster, see [Changing an Amazon MSK cluster's security group](https://docs.aws.amazon.com/msk/latest/developerguide/change-security-group.html).
AWS Database Migration Service publishes records to a Kafka topic using JSON. During conversion, AWS DMS serializes each record from the source database into an attribute-value pair in JSON format.
To migrate your data from any supported data source to a target Kafka cluster, you use object mapping. With object mapping, you determine how to structure the data records in the target topic. You also define a partition key for each table, which Apache Kafka uses to group the data into its partitions.
Currently, AWS DMS supports a single topic per task. For a single task with multiple tables, all messages go to a single topic. Each message includes a metadata section that identifies the target schema and table. AWS DMS versions 3.4.6 and higher support multitopic replication using object mapping. For more information, see [Multitopic replication using object mapping](#CHAP_Target.Kafka.MultiTopic).
**Apache Kafka endpoint settings**
You can specify connection details through endpoint settings in the AWS DMS console, or the `--kafka-settings` option in the CLI. The requirements for each setting follow:
+ `Broker` – Specify the locations of one or more brokers in your Kafka cluster in the form of a comma-separated list of each `broker-hostname:port`. An example is `"ec2-12-345-678-901.compute-1.amazonaws.com:2345,ec2-10-987-654-321.compute-1.amazonaws.com:9876"`. This setting can specify the locations of any or all brokers in the cluster. The cluster brokers all communicate to handle the partitioning of data records migrated to the topic.
+ `Topic` – (Optional) Specify the topic name with a maximum length of 255 letters and symbols. You can use period (.), underscore (\$1), and minus (-). Topic names with a period (.) or underscore (\$1) can collide in internal data structures. Use either one, but not both of these symbols in the topic name. If you don't specify a topic name, AWS DMS uses `"kafka-default-topic"` as the migration topic.
**Note**
To have AWS DMS create either a migration topic you specify or the default topic, set `auto.create.topics.enable = true` as part of your Kafka cluster configuration. For more information, see [Limitations when using Apache Kafka as a target for AWS Database Migration Service](#CHAP_Target.Kafka.Limitations)
+ `MessageFormat` – The output format for the records created on the endpoint. The message format is `JSON` (default) or `JSON_UNFORMATTED` (a single line with no tab).
+ `MessageMaxBytes` – The maximum size in bytes for records created on the endpoint. The default is 1,000,000.
**Note**
You can only use the AWS CLI/SDK to change `MessageMaxBytes` to a non-default value. For example, to modify your existing Kafka endpoint and change `MessageMaxBytes`, use the following command.
```
aws dms modify-endpoint --endpoint-arn your-endpoint
--kafka-settings Broker="broker1-server:broker1-port,broker2-server:broker2-port,...",
Topic=topic-name,MessageMaxBytes=integer-of-max-message-size-in-bytes
```
+ `IncludeTransactionDetails` – Provides detailed transaction information from the source database. This information includes a commit timestamp, a log position, and values for `transaction_id`, `previous_transaction_id`, and `transaction_record_id` (the record offset within a transaction). The default is `false`.
+ `IncludePartitionValue` – Shows the partition value within the Kafka message output, unless the partition type is `schema-table-type`. The default is `false`.
+ `PartitionIncludeSchemaTable` – Prefixes schema and table names to partition values, when the partition type is `primary-key-type`. Doing this increases data distribution among Kafka partitions. For example, suppose that a `SysBench` schema has thousands of tables and each table has only limited range for a primary key. In this case, the same primary key is sent from thousands of tables to the same partition, which causes throttling. The default is `false`.
+ `IncludeTableAlterOperations` – Includes any data definition language (DDL) operations that change the table in the control data, such as `rename-table`, `drop-table`, `add-column`, `drop-column`, and `rename-column`. The default is `false`.
+ `IncludeControlDetails` – Shows detailed control information for table definition, column definition, and table and column changes in the Kafka message output. The default is `false`.
+ `IncludeNullAndEmpty` – Include NULL and empty columns in the target. The default is `false`.
+ `SecurityProtocol` – Sets a secure connection to a Kafka target endpoint using Transport Layer Security (TLS). Options include `ssl-authentication`, `ssl-encryption`, and `sasl-ssl`. Using `sasl-ssl` requires `SaslUsername` and `SaslPassword`.
+ `SslEndpointIdentificationAlgorithm` – Sets hostname verification for the certificate. This setting is supported in AWS DMS version 3.5.1 and later. Options include the following:
+ `NONE`: Disable hostname verification of the broker in the client connection.
+ `HTTPS`: Enable hostname verification of the broker in the client connection.
+ `useLargeIntegerValue` – Use up to 18 digit int instead of casting ints as doubles, available from AWS DMS version 3.5.4. The default is false.
You can use settings to help increase the speed of your transfer. To do so, AWS DMS supports a multithreaded full load to an Apache Kafka target cluster. AWS DMS supports this multithreading with task settings that include the following:
+ `MaxFullLoadSubTasks` – Use this option to indicate the maximum number of source tables to load in parallel. AWS DMS loads each table into its corresponding Kafka target table using a dedicated subtask. The default is 8; the maximum value is 49.
+ `ParallelLoadThreads` – Use this option to specify the number of threads that AWS DMS uses to load each table into its Kafka target table. The maximum value for an Apache Kafka target is 32. You can ask to have this maximum limit increased.
+ `ParallelLoadBufferSize` – Use this option to specify the maximum number of records to store in the buffer that the parallel load threads use to load data to the Kafka target. The default value is 50. The maximum value is 1,000. Use this setting with `ParallelLoadThreads`. `ParallelLoadBufferSize` is valid only when there is more than one thread.
+ `ParallelLoadQueuesPerThread` – Use this option to specify the number of queues each concurrent thread accesses to take data records out of queues and generate a batch load for the target. The default is 1. The maximum is 512.
You can improve the performance of change data capture (CDC) for Kafka endpoints by tuning task settings for parallel threads and bulk operations. To do this, you can specify the number of concurrent threads, queues per thread, and the number of records to store in a buffer using `ParallelApply*` task settings. For example, suppose you want to perform a CDC load and apply 128 threads in parallel. You also want to access 64 queues per thread, with 50 records stored per buffer.
To promote CDC performance, AWS DMS supports these task settings:
+ `ParallelApplyThreads` – Specifies the number of concurrent threads that AWS DMS uses during a CDC load to push data records to a Kafka target endpoint. The default value is zero (0) and the maximum value is 32.
+ `ParallelApplyBufferSize` – Specifies the maximum number of records to store in each buffer queue for concurrent threads to push to a Kafka target endpoint during a CDC load. The default value is 100 and the maximum value is 1,000. Use this option when `ParallelApplyThreads` specifies more than one thread.
+ `ParallelApplyQueuesPerThread` – Specifies the number of queues that each thread accesses to take data records out of queues and generate a batch load for a Kafka endpoint during CDC. The default is 1. The maximum is 512.
When using `ParallelApply*` task settings, the `partition-key-type` default is the `primary-key` of the table, not `schema-name.table-name`.
## Connecting to Kafka using Transport Layer Security (TLS)
A Kafka cluster accepts secure connections using Transport Layer Security (TLS). With DMS, you can use any one of the following three security protocol options to secure a Kafka endpoint connection.
**SSL encryption (`server-encryption`)**
Clients validate server identity through the server’s certificate. Then an encrypted connection is made between server and client.
**SSL authentication (`mutual-authentication`)**
Server and client validate the identity with each other through their own certificates. Then an encrypted connection is made between server and client.
**SASL-SSL (`mutual-authentication`)**
The Simple Authentication and Security Layer (SASL) method replaces the client’s certificate with a user name and password to validate a client identity. Specifically, you provide a user name and password that the server has registered so that the server can validate the identity of a client. Then an encrypted connection is made between server and client.
**Important**
Apache Kafka and Amazon MSK accept resolved certificates. This is a known limitation of Kafka and Amazon MSK to be addressed. For more information, see [Apache Kafka issues, KAFKA-3700](https://issues.apache.org/jira/browse/KAFKA-3700).
If you're using Amazon MSK, consider using access control lists (ACLs) as a workaround to this known limitation. For more information about using ACLs, see [Apache Kafka ACLs](https://docs.aws.amazon.com//msk/latest/developerguide/msk-acls.html) section of *Amazon Managed Streaming for Apache Kafka Developer Guide*.
If you're using a self-managed Kafka cluster, see [Comment dated 21/Oct/18](https://issues.apache.org/jira/browse/KAFKA-3700?focusedCommentId=16658376) for information about configuring your cluster.
### Using SSL encryption with Amazon MSK or a self-managed Kafka cluster
You can use SSL encryption to secure an endpoint connection to Amazon MSK or a self-managed Kafka cluster. When you use the SSL encryption authentication method, clients validate a server's identity through the server’s certificate. Then an encrypted connection is made between server and client.
**To use SSL encryption to connect to Amazon MSK**
+ Set the security protocol endpoint setting (`SecurityProtocol`) using the `ssl-encryption` option when you create your target Kafka endpoint.
The JSON example following sets the security protocol as SSL encryption.
```
"KafkaSettings": {
"SecurityProtocol": "ssl-encryption",
}
```
**To use SSL encryption for a self-managed Kafka cluster**
1. If you're using a private Certification Authority (CA) in your on-premises Kafka cluster, upload your private CA cert and get an Amazon Resource Name (ARN).
1. Set the security protocol endpoint setting (`SecurityProtocol`) using the `ssl-encryption` option when you create your target Kafka endpoint. The JSON example following sets the security protocol as `ssl-encryption`.
```
"KafkaSettings": {
"SecurityProtocol": "ssl-encryption",
}
```
1. If you're using a private CA, set `SslCaCertificateArn` in the ARN you got in the first step above.
### Using SSL authentication
You can use SSL authentication to secure an endpoint connection to Amazon MSK or a self-managed Kafka cluster.
To enable client authentication and encryption using SSL authentication to connect to Amazon MSK, do the following:
+ Prepare a private key and public certificate for Kafka.
+ Upload certificates to the DMS certificate manager.
+ Create a Kafka target endpoint with corresponding certificate ARNs specified in Kafka endpoint settings.
**To prepare a private key and public certificate for Amazon MSK**
1. Create an EC2 instance and set up a client to use authentication as described in steps 1 through 9 in the [Client Authentication](https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html) section of *Amazon Managed Streaming for Apache Kafka Developer Guide*.
After you complete those steps, you have a Certificate-ARN (the public certificate ARN saved in ACM), and a private key contained within a `kafka.client.keystore.jks` file.
1. Get the public certificate and copy the certificate to the `signed-certificate-from-acm.pem` file, using the command following:
```
aws acm-pca get-certificate --certificate-authority-arn Private_CA_ARN --certificate-arn Certificate_ARN
```
That command returns information similar to the following example:
```
{"Certificate": "123", "CertificateChain": "456"}
```
You then copy your equivalent of `"123"` to the `signed-certificate-from-acm.pem` file.
1. Get the private key by importing the `msk-rsa` key from `kafka.client.keystore.jks to keystore.p12`, as shown in the following example.
```
keytool -importkeystore \
-srckeystore kafka.client.keystore.jks \
-destkeystore keystore.p12 \
-deststoretype PKCS12 \
-srcalias msk-rsa-client \
-deststorepass test1234 \
-destkeypass test1234
```
1. Use the following command to export `keystore.p12` into `.pem` format.
```
Openssl pkcs12 -in keystore.p12 -out encrypted-private-client-key.pem –nocerts
```
The **Enter PEM pass phrase** message appears and identifies the key that is applied to encrypt the certificate.
1. Remove bag attributes and key attributes from the `.pem` file to make sure that the first line starts with the following string.
```
---BEGIN ENCRYPTED PRIVATE KEY---
```
**To upload a public certificate and private key to the DMS certificate manager and test the connection to Amazon MSK**
1. Upload to DMS certificate manager using the following command.
```
aws dms import-certificate --certificate-identifier signed-cert --certificate-pem file://path to signed cert
aws dms import-certificate --certificate-identifier private-key —certificate-pem file://path to private key
```
1. Create an Amazon MSK target endpoint and test connection to make sure that TLS authentication works.
```
aws dms create-endpoint --endpoint-identifier $endpoint-identifier --engine-name kafka --endpoint-type target --kafka-settings
'{"Broker": "b-0.kafka260.aaaaa1.a99.kafka.us-east-1.amazonaws.com:0000", "SecurityProtocol":"ssl-authentication",
"SslClientCertificateArn": "arn:aws:dms:us-east-1:012346789012:cert:",
"SslClientKeyArn": "arn:aws:dms:us-east-1:0123456789012:cert:","SslClientKeyPassword":"test1234"}'
aws dms test-connection -replication-instance-arn=$rep_inst_arn —endpoint-arn=$kafka_tar_arn_msk
```
**Important**
You can use SSL authentication to secure a connection to a self-managed Kafka cluster. In some cases, you might use a private Certification Authority (CA) in your on-premises Kafka cluster. If so, upload your CA chain, public certificate, and private key to the DMS certificate manager. Then, use the corresponding Amazon Resource Name (ARN) in your endpoint settings when you create your on-premises Kafka target endpoint.
**To prepare a private key and signed certificate for a self-managed Kafka cluster**
1. Generate a key pair as shown in the following example.
```
keytool -genkey -keystore kafka.server.keystore.jks -validity 300 -storepass your-keystore-password
-keypass your-key-passphrase -dname "CN=your-cn-name"
-alias alias-of-key-pair -storetype pkcs12 -keyalg RSA
```
1. Generate a Certificate Sign Request (CSR).
```
keytool -keystore kafka.server.keystore.jks -certreq -file server-cert-sign-request-rsa -alias on-premise-rsa -storepass your-key-store-password
-keypass your-key-password
```
1. Use the CA in your cluster truststore to sign the CSR. If you don't have a CA, you can create your own private CA.
```
openssl req -new -x509 -keyout ca-key -out ca-cert -days validate-days
```
1. Import `ca-cert` into the server truststore and keystore. If you don't have a truststore, use the following command to create the truststore and import `ca-cert `into it.
```
keytool -keystore kafka.server.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.server.keystore.jks -alias CARoot -import -file ca-cert
```
1. Sign the certificate.
```
openssl x509 -req -CA ca-cert -CAkey ca-key -in server-cert-sign-request-rsa -out signed-server-certificate.pem
-days validate-days -CAcreateserial -passin pass:ca-password
```
1. Import the signed certificate to the keystore.
```
keytool -keystore kafka.server.keystore.jks -import -file signed-certificate.pem -alias on-premise-rsa -storepass your-keystore-password
-keypass your-key-password
```
1. Use the following command to import the `on-premise-rsa` key from `kafka.server.keystore.jks` to `keystore.p12`.
```
keytool -importkeystore \
-srckeystore kafka.server.keystore.jks \
-destkeystore keystore.p12 \
-deststoretype PKCS12 \
-srcalias on-premise-rsa \
-deststorepass your-truststore-password \
-destkeypass your-key-password
```
1. Use the following command to export `keystore.p12` into `.pem` format.
```
Openssl pkcs12 -in keystore.p12 -out encrypted-private-server-key.pem –nocerts
```
1. Upload `encrypted-private-server-key.pem`, `signed-certificate.pem`, and `ca-cert` to the DMS certificate manager.
1. Create an endpoint by using the returned ARNs.
```
aws dms create-endpoint --endpoint-identifier $endpoint-identifier --engine-name kafka --endpoint-type target --kafka-settings
'{"Broker": "b-0.kafka260.aaaaa1.a99.kafka.us-east-1.amazonaws.com:9092", "SecurityProtocol":"ssl-authentication",
"SslClientCertificateArn": "your-client-cert-arn","SslClientKeyArn": "your-client-key-arn","SslClientKeyPassword":"your-client-key-password",
"SslCaCertificateArn": "your-ca-certificate-arn"}'
aws dms test-connection -replication-instance-arn=$rep_inst_arn —endpoint-arn=$kafka_tar_arn_msk
```
### Using SASL-SSL authentication to connect to Amazon MSK
The Simple Authentication and Security Layer (SASL) method uses a user name and password to validate a client identity, and makes an encrypted connection between server and client.
To use SASL, you first create a secure user name and password when you set up your Amazon MSK cluster. For a description how to set up a secure user name and password for an Amazon MSK cluster, see [Setting up SASL/SCRAM authentication for an Amazon MSK cluster](https://docs.aws.amazon.com/msk/latest/developerguide/msk-password.html#msk-password-tutorial) in the *Amazon Managed Streaming for Apache Kafka Developer Guide*.
Then, when you create your Kafka target endpoint, set the security protocol endpoint setting (`SecurityProtocol`) using the `sasl-ssl` option. You also set `SaslUsername` and `SaslPassword` options. Make sure these are consistent with the secure user name and password that you created when you first set up your Amazon MSK cluster, as shown in the following JSON example.
```
"KafkaSettings": {
"SecurityProtocol": "sasl-ssl",
"SaslUsername":"Amazon MSK cluster secure user name",
"SaslPassword":"Amazon MSK cluster secure password"
}
```
**Note**
Currently, AWS DMS supports only public CA backed SASL-SSL. DMS does not support SASL-SSL for use with self-managed Kafka that is backed by private CA.
For SASL-SSL authentication, AWS DMS supports the SCRAM-SHA-512 mechanism by default. AWS DMS versions 3.5.0 and higher also support the Plain mechanism. To support the Plain mechanism, set the `SaslMechanism` parameter of the `KafkaSettings` API data type to `PLAIN`. The datatype `PLAIN` is supported by Kafka, but not supported by Amazon Kafka (MSK).
## Using a before image to view original values of CDC rows for Apache Kafka as a target
When writing CDC updates to a data-streaming target like Kafka you can view a source database row's original values before change by an update. To make this possible, AWS DMS populates a *before image* of update events based on data supplied by the source database engine.
Different source database engines provide different amounts of information for a before image:
+ Oracle provides updates to columns only if they change.
+ PostgreSQL provides only data for columns that are part of the primary key (changed or not). If logical replication is in use and REPLICA IDENTITY FULL is set for the source table, you can get entire before and after information on the row written to the WALs and available here.
+ MySQL generally provides data for all columns (changed or not).
To enable before imaging to add original values from the source database to the AWS DMS output, use either the `BeforeImageSettings` task setting or the `add-before-image-columns` parameter. This parameter applies a column transformation rule.
`BeforeImageSettings` adds a new JSON attribute to every update operation with values collected from the source database system, as shown following.
```
"BeforeImageSettings": {
"EnableBeforeImage": boolean,
"FieldName": string,
"ColumnFilter": pk-only (default) / non-lob / all (but only one)
}
```
**Note**
Apply `BeforeImageSettings` to full load plus CDC tasks (which migrate existing data and replicate ongoing changes), or to CDC only tasks (which replicate data changes only). Don't apply `BeforeImageSettings` to tasks that are full load only.
For `BeforeImageSettings` options, the following applies:
+ Set the `EnableBeforeImage` option to `true` to enable before imaging. The default is `false`.
+ Use the `FieldName` option to assign a name to the new JSON attribute. When `EnableBeforeImage` is `true`, `FieldName` is required and can't be empty.
+ The `ColumnFilter` option specifies a column to add by using before imaging. To add only columns that are part of the table's primary keys, use the default value, `pk-only`. To add only columns that are not of LOB type, use `non-lob`. To add any column that has a before image value, use `all`.
```
"BeforeImageSettings": {
"EnableBeforeImage": true,
"FieldName": "before-image",
"ColumnFilter": "pk-only"
}
```
### Using a before image transformation rule
As as an alternative to task settings, you can use the `add-before-image-columns` parameter, which applies a column transformation rule. With this parameter, you can enable before imaging during CDC on data streaming targets like Kafka.
By using `add-before-image-columns` in a transformation rule, you can apply more fine-grained control of the before image results. Transformation rules enable you to use an object locator that gives you control over tables selected for the rule. Also, you can chain transformation rules together, which allows different rules to be applied to different tables. You can then manipulate the columns produced by using other rules.
**Note**
Don't use the `add-before-image-columns` parameter together with the `BeforeImageSettings` task setting within the same task. Instead, use either the parameter or the setting, but not both, for a single task.
A `transformation` rule type with the `add-before-image-columns` parameter for a column must provide a `before-image-def` section. The following shows an example.
```
{
"rule-type": "transformation",
…
"rule-target": "column",
"rule-action": "add-before-image-columns",
"before-image-def":{
"column-filter": one-of (pk-only / non-lob / all),
"column-prefix": string,
"column-suffix": string,
}
}
```
The value of `column-prefix` is prepended to a column name, and the default value of `column-prefix` is `BI_`. The value of `column-suffix` is appended to the column name, and the default is empty. Don't set both `column-prefix` and `column-suffix` to empty strings.
Choose one value for `column-filter`. To add only columns that are part of table primary keys, choose `pk-only` . Choose `non-lob` to only add columns that are not of LOB type. Or choose `all` to add any column that has a before-image value.
### Example for a before image transformation rule
The transformation rule in the following example adds a new column called `BI_emp_no` in the target. So a statement like `UPDATE employees SET emp_no = 3 WHERE emp_no = 1;` populates the `BI_emp_no` field with 1. When you write CDC updates to Amazon S3 targets, the `BI_emp_no` column makes it possible to tell which original row was updated.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "%",
"table-name": "%"
},
"rule-action": "include"
},
{
"rule-type": "transformation",
"rule-id": "2",
"rule-name": "2",
"rule-target": "column",
"object-locator": {
"schema-name": "%",
"table-name": "employees"
},
"rule-action": "add-before-image-columns",
"before-image-def": {
"column-prefix": "BI_",
"column-suffix": "",
"column-filter": "pk-only"
}
}
]
}
```
For information on using the `add-before-image-columns` rule action, see [Transformation rules and actions](CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Transformations.md).
## Limitations when using Apache Kafka as a target for AWS Database Migration Service
The following limitations apply when using Apache Kafka as a target:
+ AWS DMS Kafka target endpoints don't support IAM access control for Amazon Managed Streaming for Apache Kafka (Amazon MSK).
+ Full LOB mode is not supported.
+ Specify a Kafka configuration file for your cluster with properties that allow AWS DMS to automatically create new topics. Include the setting, `auto.create.topics.enable = true`. If you are using Amazon MSK, you can specify the default configuration when you create your Kafka cluster, then change the `auto.create.topics.enable` setting to `true`. For more information about the default configuration settings, see [The default Amazon MSK configuration](https://docs.aws.amazon.com/msk/latest/developerguide/msk-default-configuration.html) in the *Amazon Managed Streaming for Apache Kafka Developer Guide*. If you need to modify an existing Kafka cluster created using Amazon MSK, run the AWS CLI command `aws kafka create-configuration` to update your Kafka configuration, as in the following example:
```
14:38:41 $ aws kafka create-configuration --name "kafka-configuration" --kafka-versions "2.2.1" --server-properties file://~/kafka_configuration
{
"LatestRevision": {
"Revision": 1,
"CreationTime": "2019-09-06T14:39:37.708Z"
},
"CreationTime": "2019-09-06T14:39:37.708Z",
"Name": "kafka-configuration",
"Arn": "arn:aws:kafka:us-east-1:111122223333:configuration/kafka-configuration/7e008070-6a08-445f-9fe5-36ccf630ecfd-3"
}
```
Here, `//~/kafka_configuration` is the configuration file you have created with the required property settings.
If you are using your own Kafka instance installed on Amazon EC2, modify the Kafka cluster configuration with the `auto.create.topics.enable = true` setting to allow AWS DMS to automatically create new topics, using the options provided with your instance.
+ AWS DMS publishes each update to a single record in the source database as one data record (message) in a given Kafka topic regardless of transactions.
+ AWS DMS supports the following four forms for partition keys:
+ `SchemaName.TableName`: A combination of the schema and table name.
+ `${AttributeName}`: The value of one of the fields in the JSON, or the primary key of the table in the source database.
+ `transaction-id`: The CDC transaction ID. All records within the same transaction go to the same partition.
+ `constant`: A fixed literal value for every record regardless of table or data. All records are sent to the same partition key value "constant", providing strict global ordering across all tables.
```
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "TransactionIdPartitionKey",
"rule-action": "map-record-to-document",
"object-locator": {
"schema-name": "onprem",
"table-name": "it_system"
},
"mapping-parameters": {
"partition-key-type": "transaction-id | constant | attribute-name | schema-table"
}
}
```
+ The `IncludeTransactionDetails` endpoint setting is only supported when the source endpoint is Oracle, SQL Server, PostgreSQL, or MySQL. For other source endpoint types, transaction details will not be included.
+ `BatchApply` is not supported for a Kafka endpoint. Using Batch Apply (for example, the `BatchApplyEnabled` target metadata task setting) for a Kafka target might result in loss of data.
+ AWS DMS does not support migrating values of `BigInt` data type with more than 16 digits. To work around this limitation, you can use the following transformation rule to convert the `BigInt` column to a string. For more information about transformation rules, see [Transformation rules and actions](CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Transformations.md).
```
{
"rule-type": "transformation",
"rule-id": "id",
"rule-name": "name",
"rule-target": "column",
"object-locator": {
"schema-name": "valid object-mapping rule action",
"table-name": "",
"column-name": ""
},
"rule-action": "change-data-type",
"data-type": {
"type": "string",
"length": 20
}
}
```
+ AWS DMS Kafka target endpoints do not support Amazon MSK servless.
+ When defining mapping rules, having both, object mapping rule and a transformation rule is not supported. You must set only one rule.
+ AWS DMS supports SASL Authentication for Apache Kafka versions up to 3.8. If you are using Kafka 4.0 or higher, you can only connect without SASL authentication.
+ AWS DMS does not support source data containing embedded `'\0'` characters when using Kafka as a target endpoint. Data containing embedded `'\0'` characters will be truncated at the first `'\0'` character.
## Using object mapping to migrate data to a Kafka topic
AWS DMS uses table-mapping rules to map data from the source to the target Kafka topic. To map data to a target topic, you use a type of table-mapping rule called object mapping. You use object mapping to define how data records in the source map to the data records published to a Kafka topic.
Kafka topics don't have a preset structure other than having a partition key.
**Note**
You don't have to use object mapping. You can use regular table mapping for various transformations. However, the partition key type will follow these default behaviors:
Primary Key is used as a partition key for Full Load.
If no parallel-apply task settings are used, `schema.table` is used as a partition key for CDC.
If parallel-apply task settings are used, Primary key is used as a partition key for CDC.
To create an object-mapping rule, specify `rule-type` as `object-mapping`. This rule specifies what type of object mapping you want to use.
The structure for the rule is as follows.
```
{
"rules": [
{
"rule-type": "object-mapping",
"rule-id": "id",
"rule-name": "name",
"rule-action": "valid object-mapping rule action",
"object-locator": {
"schema-name": "case-sensitive schema name",
"table-name": ""
}
}
]
}
```
AWS DMS currently supports `map-record-to-record` and `map-record-to-document` as the only valid values for the `rule-action` parameter. These settings affect values that aren't excluded as part of the `exclude-columns` attribute list. The `map-record-to-record` and `map-record-to-document` values specify how AWS DMS handles these records by default. These values don't affect the attribute mappings in any way.
Use `map-record-to-record` when migrating from a relational database to a Kafka topic. This rule type uses the `taskResourceId.schemaName.tableName` value from the relational database as the partition key in the Kafka topic and creates an attribute for each column in the source database.
When using `map-record-to-record`, note the following:
+ This setting only affects columns excluded by the `exclude-columns` list.
+ For every such column, AWS DMS creates a corresponding attribute in the target topic.
+ AWS DMS creates this corresponding attribute regardless of whether the source column is used in an attribute mapping.
One way to understand `map-record-to-record` is to see it in action. For this example, assume that you are starting with a relational database table row with the following structure and data.
| FirstName | LastName | StoreId | HomeAddress | HomePhone | WorkAddress | WorkPhone | DateofBirth |
| --- | --- | --- | --- | --- | --- | --- | --- |
| Randy | Marsh | 5 | 221B Baker Street | 1234567890 | 31 Spooner Street, Quahog | 9876543210 | 02/29/1988 |
To migrate this information from a schema named `Test` to a Kafka topic, you create rules to map the data to the target topic. The following rule illustrates the mapping.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"rule-action": "include",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
}
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "DefaultMapToKafka",
"rule-action": "map-record-to-record",
"object-locator": {
"schema-name": "Test",
"table-name": "Customers"
}
}
]
}
```
Given a Kafka topic and a partition key (in this case, `taskResourceId.schemaName.tableName`), the following illustrates the resulting record format using our sample data in the Kafka target topic:
```
{
"FirstName": "Randy",
"LastName": "Marsh",
"StoreId": "5",
"HomeAddress": "221B Baker Street",
"HomePhone": "1234567890",
"WorkAddress": "31 Spooner Street, Quahog",
"WorkPhone": "9876543210",
"DateOfBirth": "02/29/1988"
}
```
**Topics**
+ [Restructuring data with attribute mapping](#CHAP_Target.Kafka.AttributeMapping)
+ [Multitopic replication using object mapping](#CHAP_Target.Kafka.MultiTopic)
+ [Message format for Apache Kafka](#CHAP_Target.Kafka.Messageformat)
### Restructuring data with attribute mapping
You can restructure the data while you are migrating it to a Kafka topic using an attribute map. For example, you might want to combine several fields in the source into a single field in the target. The following attribute map illustrates how to restructure the data.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"rule-action": "include",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
}
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "TransformToKafka",
"rule-action": "map-record-to-record",
"target-table-name": "CustomerData",
"object-locator": {
"schema-name": "Test",
"table-name": "Customers"
},
"mapping-parameters": {
"partition-key-type": "attribute-name",
"partition-key-name": "CustomerName",
"exclude-columns": [
"firstname",
"lastname",
"homeaddress",
"homephone",
"workaddress",
"workphone"
],
"attribute-mappings": [
{
"target-attribute-name": "CustomerName",
"attribute-type": "scalar",
"attribute-sub-type": "string",
"value": "${lastname}, ${firstname}"
},
{
"target-attribute-name": "ContactDetails",
"attribute-type": "document",
"attribute-sub-type": "json",
"value": {
"Home": {
"Address": "${homeaddress}",
"Phone": "${homephone}"
},
"Work": {
"Address": "${workaddress}",
"Phone": "${workphone}"
}
}
}
]
}
}
]
}
```
To set a constant value for `partition-key`, specify `"partition-key-type: "constant"`, this sets the partition value to `constant`. For example, you might do this to force all the data to be stored in a single partition. The following mapping illustrates this approach.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
},
"rule-action": "include"
},
{
"rule-type": "object-mapping",
"rule-id": "1",
"rule-name": "TransformToKafka",
"rule-action": "map-record-to-document",
"object-locator": {
"schema-name": "Test",
"table-name": "Customer"
},
"mapping-parameters": {
"partition-key-type": "constant",
"exclude-columns": [
"FirstName",
"LastName",
"HomeAddress",
"HomePhone",
"WorkAddress",
"WorkPhone"
],
"attribute-mappings": [
{
"attribute-name": "CustomerName",
"value": "${FirstName},${LastName}"
},
{
"attribute-name": "ContactDetails",
"value": {
"Home": {
"Address": "${HomeAddress}",
"Phone": "${HomePhone}"
},
"Work": {
"Address": "${WorkAddress}",
"Phone": "${WorkPhone}"
}
}
},
{
"attribute-name": "DateOfBirth",
"value": "${DateOfBirth}"
}
]
}
}
]
}
```
**Note**
The `partition-key` value for a control record that is for a specific table is `TaskId.SchemaName.TableName`. The `partition-key` value for a control record that is for a specific task is that record's `TaskId`. Specifying a `partition-key` value in the object mapping has no impact on the `partition-key` for a control record.
When `partition-key-type` is set to `attribute-name` in a table mapping rule, you must specify `partition-key-name`, which must reference either a column from the source table or a custom column defined in the mapping. Additionally, `attribute-mappings` must be provided to define how source columns map to the target Kafka topic.
### Multitopic replication using object mapping
By default, AWS DMS tasks migrate all source data to one of the Kafka topics following:
+ As specified in the **Topic** field of the AWS DMS target endpoint.
+ As specified by `kafka-default-topic` if the **Topic** field of the target endpoint isn't populated and the Kafka `auto.create.topics.enable` setting is set to `true`.
With AWS DMS engine versions 3.4.6 and higher, you can use the `kafka-target-topic` attribute to map each migrated source table to a separate topic. For example, the object mapping rules following migrate the source tables `Customer` and `Address` to the Kafka topics `customer_topic` and `address_topic`, respectively. At the same time, AWS DMS migrates all other source tables, including the `Bills` table in the `Test` schema, to the topic specified in the target endpoint.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"rule-action": "include",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
}
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "MapToKafka1",
"rule-action": "map-record-to-record",
"kafka-target-topic": "customer_topic",
"object-locator": {
"schema-name": "Test",
"table-name": "Customer"
},
"partition-key-type": "constant"
},
{
"rule-type": "object-mapping",
"rule-id": "3",
"rule-name": "MapToKafka2",
"rule-action": "map-record-to-record",
"kafka-target-topic": "address_topic",
"object-locator": {
"schema-name": "Test",
"table-name": "Address"
},
"partition-key-type": "constant"
},
{
"rule-type": "object-mapping",
"rule-id": "4",
"rule-name": "DefaultMapToKafka",
"rule-action": "map-record-to-record",
"object-locator": {
"schema-name": "Test",
"table-name": "Bills"
}
}
]
}
```
By using Kafka multitopic replication, you can group and migrate source tables to separate Kafka topics using a single replication task.
### Message format for Apache Kafka
The JSON output is simply a list of key-value pairs.
**RecordType**
The record type can be either data or control. *Data records *represent the actual rows in the source. *Control records* are for important events in the stream, for example a restart of the task.
**Operation**
For data records, the operation can be `load`, `insert`, `update`, or `delete`.
For control records, the operation can be `create-table`, `rename-table`, `drop-table`, `change-columns`, `add-column`, `drop-column`, `rename-column`, or `column-type-change`.
**SchemaName**
The source schema for the record. This field can be empty for a control record.
**TableName**
The source table for the record. This field can be empty for a control record.
**Timestamp**
The timestamp for when the JSON message was constructed. The field is formatted with the ISO 8601 format.
The following JSON message example illustrates a data type message with all additional metadata.
```
{
"data":{
"id":100000161,
"fname":"val61s",
"lname":"val61s",
"REGION":"val61s"
},
"metadata":{
"timestamp":"2019-10-31T22:53:59.721201Z",
"record-type":"data",
"operation":"insert",
"partition-key-type":"primary-key",
"partition-key-value":"sbtest.sbtest_x.100000161",
"schema-name":"sbtest",
"table-name":"sbtest_x",
"transaction-id":9324410911751,
"transaction-record-id":1,
"prev-transaction-id":9324410910341,
"prev-transaction-record-id":10,
"commit-timestamp":"2019-10-31T22:53:55.000000Z",
"stream-position":"mysql-bin-changelog.002171:36912271:0:36912333:9324410911751:mysql-bin-changelog.002171:36912209"
}
}
```
The following JSON message example illustrates a control type message.
```
{
"control":{
"table-def":{
"columns":{
"id":{
"type":"WSTRING",
"length":512,
"nullable":false
},
"fname":{
"type":"WSTRING",
"length":255,
"nullable":true
},
"lname":{
"type":"WSTRING",
"length":255,
"nullable":true
},
"REGION":{
"type":"WSTRING",
"length":1000,
"nullable":true
}
},
"primary-key":[
"id"
],
"collation-name":"latin1_swedish_ci"
}
},
"metadata":{
"timestamp":"2019-11-21T19:14:22.223792Z",
"record-type":"control",
"operation":"create-table",
"partition-key-type":"task-id",
"schema-name":"sbtest",
"table-name":"sbtest_t1"
}
}
```
# Using an Amazon OpenSearch Service cluster as a target for AWS Database Migration Service
You can use AWS DMS to migrate data to Amazon OpenSearch Service (OpenSearch Service). OpenSearch Service is a managed service that makes it easy to deploy, operate, and scale an OpenSearch Service cluster.
In OpenSearch Service, you work with indexes and documents. An *index* is a collection of documents, and a *document* is a JSON object containing scalar values, arrays, and other objects. OpenSearch provides a JSON-based query language, so that you can query data in an index and retrieve the corresponding documents.
When AWS DMS creates indexes for a target endpoint for OpenSearch Service, it creates one index for each table from the source endpoint. The cost for creating an OpenSearch Service index depends on several factors. These are the number of indexes created, the total amount of data in these indexes, and the small amount of metadata that OpenSearch stores for each document.
Configure your OpenSearch Service cluster with compute and storage resources that are appropriate for the scope of your migration. We recommend that you consider the following factors, depending on the replication task you want to use:
+ For a full data load, consider the total amount of data that you want to migrate, and also the speed of the transfer.
+ For replicating ongoing changes, consider the frequency of updates, and your end-to-end latency requirements.
Also, configure the index settings on your OpenSearch cluster, paying close attention to the document count.
**Multithreaded full load task settings**
To help increase the speed of the transfer, AWS DMS supports a multithreaded full load to an OpenSearch Service target cluster. AWS DMS supports this multithreading with task settings that include the following:
+ `MaxFullLoadSubTasks` – Use this option to indicate the maximum number of source tables to load in parallel. DMS loads each table into its corresponding OpenSearch Service target index using a dedicated subtask. The default is 8; the maximum value is 49.
+ `ParallelLoadThreads` – Use this option to specify the number of threads that AWS DMS uses to load each table into its OpenSearch Service target index. The maximum value for an OpenSearch Service target is 32. You can ask to have this maximum limit increased.
**Note**
If you don't change `ParallelLoadThreads` from its default (0), AWS DMS transfers a single record at a time. This approach puts undue load on your OpenSearch Service cluster. Make sure that you set this option to 1 or more.
+ `ParallelLoadBufferSize` – Use this option to specify the maximum number of records to store in the buffer that the parallel load threads use to load data to the OpenSearch Service target. The default value is 50. The maximum value is 1,000. Use this setting with `ParallelLoadThreads`. `ParallelLoadBufferSize` is valid only when there is more than one thread.
For more information on how DMS loads an OpenSearch Service cluster using multithreading, see the AWS blog post [Scale Amazon OpenSearch Service for AWS Database Migration Service migrations](https://aws.amazon.com/blogs/database/scale-amazon-elasticsearch-service-for-aws-database-migration-service-migrations/).
**Multithreaded CDC load task settings**
You can improve the performance of change data capture (CDC) for an OpenSearch Service target cluster using task settings to modify the behavior of the `PutRecords` API call. To do this, you can specify the number of concurrent threads, queues per thread, and the number of records to store in a buffer using `ParallelApply*` task settings. For example, suppose you want to perform a CDC load and apply 32 threads in parallel. You also want to access 64 queues per thread, with 50 records stored per buffer.
**Note**
Support for the use of `ParallelApply*` task settings during CDC to Amazon OpenSearch Service target endpoints is available in AWS DMS versions 3.4.0 and higher.
To promote CDC performance, AWS DMS supports these task settings:
+ `ParallelApplyThreads` – Specifies the number of concurrent threads that AWS DMS uses during a CDC load to push data records to a OpenSearch Service target endpoint. The default value is zero (0) and the maximum value is 32.
+ `ParallelApplyBufferSize` – Specifies the maximum number of records to store in each buffer queue for concurrent threads to push to a OpenSearch Service target endpoint during a CDC load. The default value is 100 and the maximum value is 1,000. Use this option when `ParallelApplyThreads` specifies more than one thread.
+ `ParallelApplyQueuesPerThread` – Specifies the number of queues that each thread accesses to take data records out of queues and generate a batch load for a OpenSearch Service endpoint during CDC.
When using `ParallelApply*` task settings, the `partition-key-type` default is the `primary-key` of the table, not `schema-name.table-name`.
## Migrating from a relational database table to an OpenSearch Service index
AWS DMS supports migrating data to OpenSearch Service's scalar data types. When migrating from a relational database like Oracle or MySQL to OpenSearch Service, you might want to restructure how you store this data.
AWS DMS supports the following OpenSearch Service scalar data types:
+ Boolean
+ Date
+ Float
+ Int
+ String
AWS DMS converts data of type Date into type String. You can specify custom mapping to interpret these dates.
AWS DMS does not support migration of LOB data types.
## Prerequisites for using Amazon OpenSearch Service as a target for AWS Database Migration Service
Before you begin work with an OpenSearch Service database as a target for AWS DMS, make sure that you create an AWS Identity and Access Management (IAM) role. This role should let AWS DMS access the OpenSearch Service indexes at the target endpoint. The minimum set of access permissions is shown in the following IAM policy.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "1",
"Effect": "Allow",
"Principal": {
"Service": "dms.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
The role that you use for the migration to OpenSearch Service must have the following permissions.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"es:ESHttpDelete",
"es:ESHttpGet",
"es:ESHttpHead",
"es:ESHttpPost",
"es:ESHttpPut"
],
"Resource": "*"
}
]
}
```
------
In the preceding example, replace `region` with the AWS Region identifier, *`account-id`* with your AWS account ID, and `domain-name` with the name of your Amazon OpenSearch Service domain. An example is `arn:aws:es:us-west-2:123456789012:domain/my-es-domain`
## Endpoint settings when using OpenSearch Service as a target for AWS DMS
You can use endpoint settings to configure your OpenSearch Service target database similar to using extra connection attributes. You specify the settings when you create the target endpoint using the AWS DMS console, or by using the `create-endpoint` command in the [AWS CLI](https://docs.aws.amazon.com/cli/latest/reference/dms/index.html), with the `--elasticsearch-settings '{"EndpointSetting": "value", ...}'` JSON syntax.
The following table shows the endpoint settings that you can use with OpenSearch Service as a target.
| Attribute name | Valid values | Default value and description |
| --- | --- | --- |
| `FullLoadErrorPercentage` | A positive integer greater than 0 but no larger than 100. | 10 – For a full load task, this attribute determines the threshold of errors allowed before the task fails. For example, suppose that there are 1,500 rows at the source endpoint and this parameter is set to 10. Then the task fails if AWS DMS encounters more than 150 errors (10 percent of the row count) when writing to the target endpoint. |
| `ErrorRetryDuration` | A positive integer greater than 0. | 300 – If an error occurs at the target endpoint, AWS DMS retries for this many seconds. Otherwise, the task fails. |
| `UseNewMappingType` | true or false | `false`, but to work using opensearch v2.x it should be set to `true`. |
## Limitations when using Amazon OpenSearch Service as a target for AWS Database Migration Service
The following limitations apply when using Amazon OpenSearch Service as a target:
+ OpenSearch Service uses dynamic mapping (auto guess) to determine the data types to use for migrated data.
+ OpenSearch Service stores each document with a unique ID. The following is an example ID.
```
"_id": "D359F8B537F1888BC71FE20B3D79EAE6674BE7ACA9B645B0279C7015F6FF19FD"
```
Each document ID is 64 bytes long, so anticipate this as a storage requirement. For example, if you migrate 100,000 rows from an AWS DMS source, the resulting OpenSearch Service index requires storage for an additional 6,400,000 bytes.
+ With OpenSearch Service, you can't make updates to the primary key attributes. This restriction is important when using ongoing replication with change data capture (CDC) because it can result in unwanted data in the target. In CDC mode, primary keys are mapped to SHA256 values, which are 32 bytes long. These are converted to human-readable 64-byte strings, and are used as OpenSearch Service document IDs.
+ If AWS DMS encounters any items that can't be migrated, it writes error messages to Amazon CloudWatch Logs. This behavior differs from that of other AWS DMS target endpoints, which write errors to an exceptions table.
+ AWS DMS does not support connection to an Amazon ES cluster that has Fine-grained Access Control enabled with master user and password.
+ AWS DMS does not support OpenSearch Service serverless.
+ OpenSearch Service does not support writing data to pre-existing indexes.
+ The replication task setting, `TargetTablePrepMode:TRUNCATE_BEFORE_LOAD` is not supported for use with a OpenSearch target endpoint.
+ When migrating data to Amazon Elasticsearch using AWS DMS, the source data must have a primary key or a unique identifier column. If the source data does not have a primary key or unique identifier, you need to define one using the define-primary-key transformation rule.
## Target data types for Amazon OpenSearch Service
When AWS DMS migrates data from heterogeneous databases, the service maps data types from the source database to intermediate data types called AWS DMS data types. The service then maps the intermediate data types to the target data types. The following table shows each AWS DMS data type and the data type it maps to in OpenSearch Service.
| AWS DMS data type | OpenSearch Service data type |
| --- | --- |
| Boolean | boolean |
| Date | string |
| Time | date |
| Timestamp | date |
| INT4 | integer |
| Real4 | float |
| UINT4 | integer |
For additional information about AWS DMS data types, see [Data types for AWS Database Migration Service](CHAP_Reference.DataTypes.md).
# Using Amazon DocumentDB as a target for AWS Database Migration Service
For information about what versions of Amazon DocumentDB (with MongoDB compatibility) that AWS DMS supports, see [Targets for AWS DMS](CHAP_Introduction.Targets.md). You can use AWS DMS to migrate data to Amazon DocumentDB (with MongoDB compatibility) from any of the source data engines that AWS DMS supports. The source engine can be on an AWS managed service such as Amazon RDS, Aurora, or Amazon S3. Or the engine can be on a self-managed database, such as MongoDB running on Amazon EC2 or on-premises.
You can use AWS DMS to replicate source data to Amazon DocumentDB databases, collections, or documents.
**Note**
If your source endpoint is MongoDB or Amazon DocumentDB, run the migration in **Document mode**.
MongoDB stores data in a binary JSON format (BSON). AWS DMS supports all of the BSON data types that are supported by Amazon DocumentDB. For a list of these data types, see [Supported MongoDB APIs, operations, and data types](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html) in *the Amazon DocumentDB Developer Guide.*
If the source endpoint is a relational database, AWS DMS maps database objects to Amazon DocumentDB as follows:
+ A relational database, or database schema, maps to an Amazon DocumentDB *database*.
+ Tables within a relational database map to *collections* in Amazon DocumentDB.
+ Records in a relational table map to *documents* in Amazon DocumentDB. Each document is constructed from data in the source record.
If the source endpoint is Amazon S3, then the resulting Amazon DocumentDB objects correspond to AWS DMS mapping rules for Amazon S3. For example, consider the following URI.
```
s3://amzn-s3-demo-bucket/hr/employee
```
In this case, AWS DMS maps the objects in `amzn-s3-demo-bucket` to Amazon DocumentDB as follows:
+ The top-level URI part (`hr`) maps to an Amazon DocumentDB database.
+ The next URI part (`employee`) maps to an Amazon DocumentDB collection.
+ Each object in `employee` maps to a document in Amazon DocumentDB.
For more information on mapping rules for Amazon S3, see [Using Amazon S3 as a source for AWS DMS](CHAP_Source.S3.md).
**Amazon DocumentDB endpoint settings**
In AWS DMS versions 3.5.0 and higher, you can improve the performance of change data capture (CDC) for Amazon DocumentDB endpoints by tuning task settings for parallel threads and bulk operations. To do this, you can specify the number of concurrent threads, queues per thread, and the number of records to store in a buffer using `ParallelApply*` task settings. For example, suppose you want to perform a CDC load and apply 128 threads in parallel. You also want to access 64 queues per thread, with 50 records stored per buffer.
To promote CDC performance, AWS DMS supports these task settings:
+ `ParallelApplyThreads` – Specifies the number of concurrent threads that AWS DMS uses during a CDC load to push data records to a Amazon DocumentDB target endpoint. The default value is zero (0) and the maximum value is 32.
+ `ParallelApplyBufferSize` – Specifies the maximum number of records to store in each buffer queue for concurrent threads to push to a Amazon DocumentDB target endpoint during a CDC load. The default value is 100 and the maximum value is 1,000. Use this option when `ParallelApplyThreads` specifies more than one thread.
+ `ParallelApplyQueuesPerThread` – Specifies the number of queues that each thread accesses to take data records out of queues and generate a batch load for a Amazon DocumentDB endpoint during CDC. The default is 1. The maximum is 512.
**Note**
For Amazon DocumentDB targets, parallel CDC apply can cause duplicate key errors or stalled CDC apply for workloads that use secondary unique indexes or require strict ordering of changes. Use the default single-threaded CDC apply configuration for these workloads.
For additional details on working with Amazon DocumentDB as a target for AWS DMS, see the following sections:
**Topics**
+ [Mapping data from a source to an Amazon DocumentDB target](#CHAP_Target.DocumentDB.data-mapping)
+ [Connecting to Amazon DocumentDB Elastic Clusters as a target](#CHAP_Target.DocumentDB.data-mapping.elastic-cluster-connect)
+ [Ongoing replication with Amazon DocumentDB as a target](#CHAP_Target.DocumentDB.data-mapping.ongoing-replication)
+ [Limitations to using Amazon DocumentDB as a target](#CHAP_Target.DocumentDB.limitations)
+ [Using endpoint settings with Amazon DocumentDB as a target](#CHAP_Target.DocumentDB.ECAs)
+ [Target data types for Amazon DocumentDB](#CHAP_Target.DocumentDB.datatypes)
**Note**
For a step-by-step walkthrough of the migration process, see [Migrating from MongoDB to Amazon DocumentDB ](https://docs.aws.amazon.com/dms/latest/sbs/CHAP_MongoDB2DocumentDB.html) in the AWS Database Migration Service Step-by-Step Migration Guide.
## Mapping data from a source to an Amazon DocumentDB target
AWS DMS reads records from the source endpoint, and constructs JSON documents based on the data it reads. For each JSON document, AWS DMS must determine an `_id` field to act as a unique identifier. It then writes the JSON document to an Amazon DocumentDB collection, using the `_id` field as a primary key.
### Source data that is a single column
If the source data consists of a single column, the data must be of a string type. (Depending on the source engine, the actual data type might be VARCHAR, NVARCHAR, TEXT, LOB, CLOB, or similar.) AWS DMS assumes that the data is a valid JSON document, and replicates the data to Amazon DocumentDB as is.
If the resulting JSON document contains a field named `_id`, then that field is used as the unique `_id` in Amazon DocumentDB.
If the JSON doesn't contain an `_id` field, then Amazon DocumentDB generates an `_id` value automatically.
### Source data that is multiple columns
If the source data consists of multiple columns, then AWS DMS constructs a JSON document from all of these columns. To determine the `_id` field for the document, AWS DMS proceeds as follows:
+ If one of the columns is named `_id`, then the data in that column is used as the target`_id`.
+ If there is no `_id` column, but the source data has a primary key or a unique index, then AWS DMS uses that key or index value as the `_id` value. The data from the primary key or unique index also appears as explicit fields in the JSON document.
+ If there is no `_id` column, and no primary key or a unique index, then Amazon DocumentDB generates an `_id` value automatically.
### Coercing a data type at the target endpoint
AWS DMS can modify data structures when it writes to an Amazon DocumentDB target endpoint. You can request these changes by renaming columns and tables at the source endpoint, or by providing transformation rules that are applied when a task is running.
#### Using a nested JSON document (json\$1 prefix)
To coerce a data type, you can prefix the source column name with `json_` (that is, `json_columnName`) either manually or using a transformation. In this case, the column is created as a nested JSON document within the target document, rather than as a string field.
For example, suppose that you want to migrate the following document from a MongoDB source endpoint.
```
{
"_id": "1",
"FirstName": "John",
"LastName": "Doe",
"ContactDetails": "{"Home": {"Address": "Boston","Phone": "1111111"},"Work": { "Address": "Boston", "Phone": "2222222222"}}"
}
```
If you don't coerce any of the source data types, the embedded `ContactDetails` document is migrated as a string.
```
{
"_id": "1",
"FirstName": "John",
"LastName": "Doe",
"ContactDetails": "{\"Home\": {\"Address\": \"Boston\",\"Phone\": \"1111111\"},\"Work\": { \"Address\": \"Boston\", \"Phone\": \"2222222222\"}}"
}
```
However, you can add a transformation rule to coerce `ContactDetails` to a JSON object. For example, suppose that the original source column name is `ContactDetails`. To coerce the data type as Nested JSON, the column at source endpoint needs to be renamed as json\$1ContactDetails” either by adding “\$1json\$1\$1“ prefix on the source manually or through transformation rules. For example, you can use the below transformation rule:
```
{
"rules": [
{
"rule-type": "transformation",
"rule-id": "1",
"rule-name": "1",
"rule-target": "column",
"object-locator": {
"schema-name": "%",
"table-name": "%",
"column-name": "ContactDetails"
},
"rule-action": "rename",
"value": "json_ContactDetails",
"old-value": null
}
]
}
```
AWS DMS replicates the ContactDetails field as nested JSON, as follows.
```
{
"_id": "1",
"FirstName": "John",
"LastName": "Doe",
"ContactDetails": {
"Home": {
"Address": "Boston",
"Phone": "1111111111"
},
"Work": {
"Address": "Boston",
"Phone": "2222222222"
}
}
}
```
#### Using a JSON array (array\$1 prefix)
To coerce a data type, you can prefix a column name with `array_` (that is, `array_columnName`), either manually or using a transformation. In this case, AWS DMS considers the column as a JSON array, and creates it as such in the target document.
Suppose that you want to migrate the following document from a MongoDB source endpoint.
```
{
"_id" : "1",
"FirstName": "John",
"LastName": "Doe",
"ContactAddresses": ["Boston", "New York"],
"ContactPhoneNumbers": ["1111111111", "2222222222"]
}
```
If you don't coerce any of the source data types, the embedded `ContactDetails` document is migrated as a string.
```
{
"_id": "1",
"FirstName": "John",
"LastName": "Doe",
"ContactAddresses": "[\"Boston\", \"New York\"]",
"ContactPhoneNumbers": "[\"1111111111\", \"2222222222\"]"
}
```
However, you can add transformation rules to coerce `ContactAddress` and `ContactPhoneNumbers` to JSON arrays, as shown in the following table.
****
| Original source column name | Renamed source column |
| --- | --- |
| ContactAddress | array\$1ContactAddress |
| ContactPhoneNumbers | array\$1ContactPhoneNumbers |
AWS DMS replicates `ContactAddress` and `ContactPhoneNumbers` as follows.
```
{
"_id": "1",
"FirstName": "John",
"LastName": "Doe",
"ContactAddresses": [
"Boston",
"New York"
],
"ContactPhoneNumbers": [
"1111111111",
"2222222222"
]
}
```
### Connecting to Amazon DocumentDB using TLS
By default, a newly created Amazon DocumentDB cluster accepts secure connections only using Transport Layer Security (TLS). When TLS is enabled, every connection to Amazon DocumentDB requires a public key.
You can retrieve the public key for Amazon DocumentDB by downloading the file, `rds-combined-ca-bundle.pem`, from an AWS hosted Amazon S3 bucket. For more information on downloading this file, see [Encrypting connections using TLS](https://docs.aws.amazon.com/documentdb/latest/developerguide/security.encryption.ssl.html) in the *Amazon DocumentDB Developer Guide*
After you download this .pem file, you can import the public key that it contains into AWS DMS as described following.
#### AWS Management Console
**To import the public key (.pem) file**
1. Open the AWS DMS console at [https://console.aws.amazon.com/dms](https://console.aws.amazon.com/dms).
1. In the navigation pane, choose **Certificates**.
1. Choose **Import certificate** and do the following:
+ For **Certificate identifier**, enter a unique name for the certificate, for example `docdb-cert`.
+ For **Import file**, navigate to the location where you saved the .pem file.
When the settings are as you want them, choose **Add new CA certificate**.
#### AWS CLI
Use the `aws dms import-certificate` command, as shown in the following example.
```
aws dms import-certificate \
--certificate-identifier docdb-cert \
--certificate-pem file://./rds-combined-ca-bundle.pem
```
When you create an AWS DMS target endpoint, provide the certificate identifier (for example, `docdb-cert`). Also, set the SSL mode parameter to `verify-full`.
## Connecting to Amazon DocumentDB Elastic Clusters as a target
In AWS DMS versions 3.4.7 and higher, you can create a Amazon DocumentDB target endpoint as an Elastic Cluster. If you create your target endpoint as an Elastic Cluster, you need to attach a new SSL certificate to your Amazon DocumentDB Elastic Cluster endpoint because your existing SSL certificate won't work.
**To attach a new SSL certificate to your Amazon DocumentDB Elastic Cluster endpoint**
1. In a browser, open [ https://www.amazontrust.com/repository/SFSRootCAG2.pem](https://www.amazontrust.com/repository/SFSRootCAG2.pem) and save the contents to a `.pem` file with a unique file name, for example `SFSRootCAG2.pem`. This is the certificate file that you need to import in subsequent steps.
1. Create the Elastic Cluster endpoint and set the following options:
1. Under **Endpoint Configuration**, choose **Add new CA certificate**.
1. For **Certificate identifier**, enter **SFSRootCAG2.pem**.
1. For **Import certificate file**, choose **Choose file**, then navigate to the `SFSRootCAG2.pem` file that you previously downloaded.
1. Select and open the downloaded `SFSRootCAG2.pem` file.
1. Choose **Import certificate**.
1. From the **Choose a certificate** drop down, choose **SFSRootCAG2.pem**.
The new SSL certificate from the downloaded `SFSRootCAG2.pem` file is now attached to your Amazon DocumentDB Elastic Cluster endpoint.
## Ongoing replication with Amazon DocumentDB as a target
If ongoing replication (change data capture, CDC) is enabled for Amazon DocumentDB as a target, AWS DMS versions 3.5.0 and higher provide a performance improvement that is twenty times greater than in prior releases. In prior releases where AWS DMS handles up to 250 records per second, AWS DMS now efficiently processes over 5000 records per second. AWS DMS also ensures that documents in Amazon DocumentDB stay in sync with the source. When a source record is created or updated, AWS DMS must first determine which Amazon DocumentDB record is affected by doing the following:
+ If the source record has a column named `_id`, the value of that column determines the corresponding `_id` in the Amazon DocumentDB collection.
+ If there is no `_id` column, but the source data has a primary key or unique index, then AWS DMS uses that key or index value as the `_id` for the Amazon DocumentDB collection.
+ If the source record doesn't have an `_id` column, a primary key, or a unique index, then AWS DMS matches all of the source columns to the corresponding fields in the Amazon DocumentDB collection.
When a new source record is created, AWS DMS writes a corresponding document to Amazon DocumentDB. If an existing source record is updated, AWS DMS updates the corresponding fields in the target document in Amazon DocumentDB. Any fields that exist in the target document but not in the source record remain untouched.
When a source record is deleted, AWS DMS deletes the corresponding document from Amazon DocumentDB.
### Structural changes (DDL) at the source
With ongoing replication, any changes to source data structures (such as tables, columns, and so on) are propagated to their counterparts in Amazon DocumentDB. In relational databases, these changes are initiated using data definition language (DDL) statements. You can see how AWS DMS propagates these changes to Amazon DocumentDB in the following table.
****
| DDL at source | Effect at Amazon DocumentDB target |
| --- | --- |
| CREATE TABLE | Creates an empty collection. |
| Statement that renames a table (RENAME TABLE, ALTER TABLE...RENAME, and similar) | Renames the collection. |
| TRUNCATE TABLE | Removes all the documents from the collection, but only if HandleSourceTableTruncated is true. For more information, see [Task settings for change processing DDL handling](CHAP_Tasks.CustomizingTasks.TaskSettings.DDLHandling.md). |
| DROP TABLE | Deletes the collection, but only if HandleSourceTableDropped is true. For more information, see [Task settings for change processing DDL handling](CHAP_Tasks.CustomizingTasks.TaskSettings.DDLHandling.md). |
| Statement that adds a column to a table (ALTER TABLE...ADD and similar) | The DDL statement is ignored, and a warning is issued. When the first INSERT is performed at the source, the new field is added to the target document. |
| ALTER TABLE...RENAME COLUMN | The DDL statement is ignored, and a warning is issued. When the first INSERT is performed at the source, the newly named field is added to the target document. |
| ALTER TABLE...DROP COLUMN | The DDL statement is ignored, and a warning is issued. |
| Statement that changes the column data type (ALTER COLUMN...MODIFY and similar) | The DDL statement is ignored, and a warning is issued. When the first INSERT is performed at the source with the new data type, the target document is created with a field of that new data type. |
## Limitations to using Amazon DocumentDB as a target
The following limitations apply when using Amazon DocumentDB as a target for AWS DMS:
+ In Amazon DocumentDB, collection names can't contain the dollar symbol (\$1). In addition, database names can't contain any Unicode characters.
+ AWS DMS doesn't support merging of multiple source tables into a single Amazon DocumentDB collection.
+ When AWS DMS processes changes from a source table that doesn't have a primary key, any LOB columns in that table are ignored.
+ If the **Change table** option is enabled and AWS DMS encounters a source column named "*\$1id*", then that column appears as "*\$1\$1id*" (two underscores) in the change table.
+ If you choose Oracle as a source endpoint, then the Oracle source must have full supplemental logging enabled. Otherwise, if there are columns at the source that weren't changed, then the data is loaded into Amazon DocumentDB as null values.
+ The replication task setting, `TargetTablePrepMode:TRUNCATE_BEFORE_LOAD` isn't supported for use with a DocumentDB target endpoint.
+ MongoDB capped collections are not supported in Amazon DocumentDB. However, AWS DMS automatically migrates such objects as uncapped collections on target DocumentDB.
+ Parallel CDC apply to Amazon DocumentDB targets can cause duplicate key errors or stalled CDC apply for workloads that use secondary unique indexes or require strict ordering of changes. For such workloads, use the default single-threaded CDC apply configuration.
## Using endpoint settings with Amazon DocumentDB as a target
You can use endpoint settings to configure your Amazon DocumentDB target database similar to using extra connection attributes. You specify the settings when you create the target endpoint using the AWS DMS console, or by using the `create-endpoint` command in the [AWS CLI](https://docs.aws.amazon.com/cli/latest/reference/dms/index.html), with the `--doc-db-settings '{"EndpointSetting": "value", ...}'` JSON syntax.
The following table shows the endpoint settings that you can use with Amazon DocumentDB as a target.
| Attribute name | Valid values | Default value and description |
| --- | --- | --- |
| `replicateShardCollections` | boolean `true` `false` | When `true`, this endpoint setting has the following effects and imposes the following limitations: [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.DocumentDB.html) |
## Target data types for Amazon DocumentDB
In the following table, you can find the Amazon DocumentDB target data types that are supported when using AWS DMS, and the default mapping from AWS DMS data types. For more information about AWS DMS data types, see [Data types for AWS Database Migration Service](CHAP_Reference.DataTypes.md).
| AWS DMS data type | Amazon DocumentDB data type |
| --- | --- |
| BOOLEAN | Boolean |
| BYTES | Binary data |
| DATE | Date |
| TIME | String (UTF8) |
| DATETIME | Date |
| INT1 | 32-bit integer |
| INT2 | 32-bit integer |
| INT4 | 32-bit integer |
| INT8 | 64-bit integer |
| NUMERIC | String (UTF8) |
| REAL4 | Double |
| REAL8 | Double |
| STRING | If the data is recognized as JSON, then AWS DMS migrates it to Amazon DocumentDB as a document. Otherwise, the data is mapped to String (UTF8). |
| UINT1 | 32-bit integer |
| UINT2 | 32-bit integer |
| UINT4 | 64-bit integer |
| UINT8 | String (UTF8) |
| WSTRING | If the data is recognized as JSON, then AWS DMS migrates it to Amazon DocumentDB as a document. Otherwise, the data is mapped to String (UTF8). |
| BLOB | Binary |
| CLOB | If the data is recognized as JSON, then AWS DMS migrates it to Amazon DocumentDB as a document. Otherwise, the data is mapped to String (UTF8). |
| NCLOB | If the data is recognized as JSON, then AWS DMS migrates it to Amazon DocumentDB as a document. Otherwise, the data is mapped to String (UTF8). |
# Using Amazon Neptune as a target for AWS Database Migration Service
Amazon Neptune is a fast, reliable, fully managed graph database service that makes it easy to build and run applications that work with highly connected datasets. The core of Neptune is a purpose-built, high-performance graph database engine. This engine is optimized for storing billions of relationships and querying the graph with milliseconds latency. Neptune supports the popular graph query languages Apache TinkerPop Gremlin and W3C's SPARQL. For more information on Amazon Neptune, see [What is Amazon Neptune?](https://docs.aws.amazon.com/neptune/latest/userguide/intro.html) in the *Amazon Neptune User Guide*.
Without a graph database such as Neptune, you probably model highly connected data in a relational database. Because the data has potentially dynamic connections, applications that use such data sources have to model connected data queries in SQL. This approach requires you to write an extra layer to convert graph queries into SQL. Also, relational databases come with schema rigidity. Any changes in the schema to model changing connections require downtime and additional maintenance of the query conversion to support the new schema. The query performance is also another big constraint to consider while designing your applications.
Graph databases can greatly simplify such situations. Free from a schema, a rich graph query layer (Gremlin or SPARQL) and indexes optimized for graph queries increase flexibility and performance. The Amazon Neptune graph database also has enterprise features such as encryption at rest, a secure authorization layer, default backups, Multi-AZ support, read replica support, and others.
Using AWS DMS, you can migrate relational data that models a highly connected graph to a Neptune target endpoint from a DMS source endpoint for any supported SQL database.
For more details, see the following.
**Topics**
+ [Overview of migrating to Amazon Neptune as a target](#CHAP_Target.Neptune.MigrationOverview)
+ [Specifying endpoint settings for Amazon Neptune as a target](#CHAP_Target.Neptune.EndpointSettings)
+ [Creating an IAM service role for accessing Amazon Neptune as a target](#CHAP_Target.Neptune.ServiceRole)
+ [Specifying graph-mapping rules using Gremlin and R2RML for Amazon Neptune as a target](#CHAP_Target.Neptune.GraphMapping)
+ [Data types for Gremlin and R2RML migration to Amazon Neptune as a target](#CHAP_Target.Neptune.DataTypes)
+ [Limitations of using Amazon Neptune as a target](#CHAP_Target.Neptune.Limitations)
## Overview of migrating to Amazon Neptune as a target
Before starting a migration to a Neptune target, create the following resources in your AWS account:
+ A Neptune cluster for the target endpoint.
+ A SQL relational database supported by AWS DMS for the source endpoint.
+ An Amazon S3 bucket for the target endpoint. Create this S3 bucket in the same AWS Region as your Neptune cluster. AWS DMS uses this S3 bucket as intermediate file storage for the target data that it bulk loads to the Neptune database. For more information on creating an S3 bucket, see [Creating a bucket](https://docs.aws.amazon.com/AmazonS3/latest/gsg/CreatingABucket.html) in the *Amazon Simple Storage Service User Guide.*
+ A virtual private cloud (VPC) endpoint for S3 in the same VPC as the Neptune cluster.
+ An AWS Identity and Access Management (IAM) role that includes an IAM policy. This policy should specify the `GetObject`, `PutObject`, `DeleteObject` and `ListObject` permissions to the S3 bucket for your target endpoint. This role is assumed by both AWS DMS and Neptune with IAM access to both the target S3 bucket and the Neptune database. For more information, see [Creating an IAM service role for accessing Amazon Neptune as a target](#CHAP_Target.Neptune.ServiceRole).
After you have these resources, setting up and starting a migration to a Neptune target is similar to any full load migration using the console or DMS API. However, a migration to a Neptune target requires some unique steps.
**To migrate an AWS DMS relational database to Neptune**
1. Create a replication instance as described in [Creating a replication instance](CHAP_ReplicationInstance.Creating.md).
1. Create and test a SQL relational database supported by AWS DMS for the source endpoint.
1. Create and test the target endpoint for your Neptune database.
To connect the target endpoint to the Neptune database, specify the server name for either the Neptune cluster endpoint or the Neptune writer instance endpoint. Also, specify the S3 bucket folder for AWS DMS to store its intermediate files for bulk load to the Neptune database.
During migration, AWS DMS stores all migrated target data in this S3 bucket folder up to a maximum file size that you specify. When this file storage reaches this maximum size, AWS DMS bulk loads the stored S3 data into the target database. It clears the folder to enable storage of any additional target data for subsequent loading to the target database. For more information on specifying these settings, see [Specifying endpoint settings for Amazon Neptune as a target](#CHAP_Target.Neptune.EndpointSettings).
1. Create a full-load replication task with the resources created in steps 1–3 and do the following:
1. Use task table mapping as usual to identify specific source schemas, tables, and views to migrate from your relational database using appropriate selection and transformation rules. For more information, see [Using table mapping to specify task settings](CHAP_Tasks.CustomizingTasks.TableMapping.md).
1. Specify target mappings by choosing one of the following to specify mapping rules from source tables and views to your Neptune target database graph:
+ Gremlin JSON – For information on using Gremlin JSON to load a Neptune database, see [Gremlin load data format](https://docs.aws.amazon.com/neptune/latest/userguide/bulk-load-tutorial-format-gremlin.html) in the *Amazon Neptune User Guide*.
+ SPARQL RDB to Resource Description Framework Mapping Language (R2RML) – For information on using SPARQL R2RML, see the W3C specification [R2RML: RDB to RDF mapping language](https://www.w3.org/TR/r2rml/).
1. Do one of the following:
+ Using the AWS DMS console, specify graph-mapping options using **Graph mapping rules** on the **Create database migration task** page.
+ Using the AWS DMS API, specify these options using the `TaskData` request parameter of the `CreateReplicationTask` API call.
For more information and examples using Gremlin JSON and SPARQL R2RML to specify graph-mapping rules, see [Specifying graph-mapping rules using Gremlin and R2RML for Amazon Neptune as a target](#CHAP_Target.Neptune.GraphMapping).
1. Start the replication for your migration task.
## Specifying endpoint settings for Amazon Neptune as a target
To create or modify a target endpoint, you can use the console or the `CreateEndpoint` or `ModifyEndpoint` API operations.
For a Neptune target in the AWS DMS console, specify **Endpoint-specific settings** on the **Create endpoint** or **Modify endpoint** console page. For `CreateEndpoint` and `ModifyEndpoint`, specify request parameters for the `NeptuneSettings` option. The following example shows how to do this using the CLI.
```
dms create-endpoint --endpoint-identifier my-neptune-target-endpoint
--endpoint-type target --engine-name neptune
--server-name my-neptune-db.cluster-cspckvklbvgf.us-east-1.neptune.amazonaws.com
--port 8192
--neptune-settings
'{"ServiceAccessRoleArn":"arn:aws:iam::123456789012:role/myNeptuneRole",
"S3BucketName":"amzn-s3-demo-bucket",
"S3BucketFolder":"amzn-s3-demo-bucket-folder",
"ErrorRetryDuration":57,
"MaxFileSize":100,
"MaxRetryCount": 10,
"IAMAuthEnabled":false}‘
```
Here, the CLI `--server-name` option specifies the server name for the Neptune cluster writer endpoint. Or you can specify the server name for a Neptune writer instance endpoint.
The `--neptune-settings` option request parameters follow:
+ `ServiceAccessRoleArn` – (Required) The Amazon Resource Name (ARN) of the service role that you created for the Neptune target endpoint. For more information, see [Creating an IAM service role for accessing Amazon Neptune as a target](#CHAP_Target.Neptune.ServiceRole).
+ `S3BucketName` – (Required) The name of the S3 bucket where DMS can temporarily store migrated graph data in .csv files before bulk loading it to the Neptune target database. DMS maps the SQL source data to graph data before storing it in these .csv files.
+ `S3BucketFolder` – (Required) A folder path where you want DMS to store migrated graph data in the S3 bucket specified by `S3BucketName`.
+ `ErrorRetryDuration` – (Optional) The number of milliseconds for DMS to wait to retry a bulk load of migrated graph data to the Neptune target database before raising an error. The default is 250.
+ `MaxFileSize` – (Optional) The maximum size in KB of migrated graph data stored in a .csv file before DMS bulk loads the data to the Neptune target database. The default is 1,048,576 KB (1 GB). If successful, DMS clears the bucket, ready to store the next batch of migrated graph data.
+ `MaxRetryCount` – (Optional) The number of times for DMS to retry a bulk load of migrated graph data to the Neptune target database before raising an error. The default is 5.
+ `IAMAuthEnabled` – (Optional) If you want IAM authorization enabled for this endpoint, set this parameter to `true` and attach the appropriate IAM policy document to your service role specified by `ServiceAccessRoleArn`. The default is `false`.
## Creating an IAM service role for accessing Amazon Neptune as a target
To access Neptune as a target, create a service role using IAM. Depending on your Neptune endpoint configuration, attach to this role some or all of the following IAM policy and trust documents. When you create the Neptune endpoint, you provide the ARN of this service role. Doing so enables AWS DMS and Amazon Neptune to assume permissions to access both Neptune and its associated Amazon S3 bucket.
If you set the `IAMAuthEnabled` parameter in `NeptuneSettings` to `true` in your Neptune endpoint configuration, attach an IAM policy like the following to your service role. If you set `IAMAuthEnabled` to `false`, you can ignore this policy.
```
// Policy to access Neptune
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": "neptune-db:*",
"Resource": "arn:aws:neptune-db:us-east-1:123456789012:cluster-CLG7H7FHK54AZGHEH6MNS55JKM/*"
}
]
}
```
The preceding IAM policy allows full access to the Neptune target cluster specified by `Resource`.
Attach an IAM policy like the following to your service role. This policy allows DMS to temporarily store migrated graph data in the S3 bucket that you created for bulk loading to the Neptune target database.
```
//Policy to access S3 bucket
{
"Version": "2012-10-17",
"Statement": [{
"Sid": "ListObjectsInBucket0",
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": [
"arn:aws:s3:::amzn-s3-demo-bucket"
]
},
{
"Sid": "AllObjectActions",
"Effect": "Allow",
"Action": ["s3:GetObject",
"s3:PutObject",
"s3:DeleteObject"
],
"Resource": [
"arn:aws:s3:::amzn-s3-demo-bucket/"
]
},
{
"Sid": "ListObjectsInBucket1",
"Effect": "Allow",
"Action": "s3:ListBucket",
"Resource": [
"arn:aws:s3:::amzn-s3-demo-bucket",
"arn:aws:s3:::amzn-s3-demo-bucket/"
]
}
]
}
```
The preceding IAM policy allows your account to query the contents of the S3 bucket (`arn:aws:s3:::amzn-s3-demo-bucket`) created for your Neptune target. It also allows your account to fully operate on the contents of all bucket files and folders (`arn:aws:s3:::amzn-s3-demo-bucket/`).
Edit the trust relationship and attach the following IAM role to your service role to allow both AWS DMS and Amazon Neptune database service to assume the role.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "",
"Effect": "Allow",
"Principal": {
"Service": "dms.amazonaws.com"
},
"Action": "sts:AssumeRole"
},
{
"Sid": "neptune",
"Effect": "Allow",
"Principal": {
"Service": "rds.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
```
------
For information about specifying this service role for your Neptune target endpoint, see [Specifying endpoint settings for Amazon Neptune as a target](#CHAP_Target.Neptune.EndpointSettings).
## Specifying graph-mapping rules using Gremlin and R2RML for Amazon Neptune as a target
The graph-mapping rules that you create specify how data extracted from an SQL relational database source is loaded into a Neptune database cluster target. The format of these mapping rules differs depending on whether the rules are for loading property-graph data using Apache TinkerPop Gremlin or Resource Description Framework (RDF) data using R2RML. Following, you can find information about these formats and where to learn more.
You can specify these mapping rules when you create the migration task using either the console or DMS API.
Using the console, specify these mapping rules using **Graph mapping rules** on the **Create database migration task** page. In **Graph mapping rules**, you can enter and edit the mapping rules directly using the editor provided. Or you can browse for a file that contains the mapping rules in the appropriate graph-mapping format.
Using the API, specify these options using the `TaskData` request parameter of the `CreateReplicationTask` API call. Set `TaskData` to the path of a file containing the mapping rules in the appropriate graph-mapping format.
### Graph-mapping rules for generating property-graph data using Gremlin
Using Gremlin to generate the property-graph data, specify a JSON object with a mapping rule for each graph entity to be generated from the source data. The format of this JSON is defined specifically for bulk loading Amazon Neptune. The following template shows what each rule in this object looks like.
```
{
"rules": [
{
"rule_id": "(an identifier for this rule)",
"rule_name": "(a name for this rule)",
"table_name": "(the name of the table or view being loaded)",
"vertex_definitions": [
{
"vertex_id_template": "{col1}",
"vertex_label": "(the vertex to create)",
"vertex_definition_id": "(an identifier for this vertex)",
"vertex_properties": [
{
"property_name": "(name of the property)",
"property_value_template": "{col2} or text",
"property_value_type": "(data type of the property)"
}
]
}
]
},
{
"rule_id": "(an identifier for this rule)",
"rule_name": "(a name for this rule)",
"table_name": "(the name of the table or view being loaded)",
"edge_definitions": [
{
"from_vertex": {
"vertex_id_template": "{col1}",
"vertex_definition_id": "(an identifier for the vertex referenced above)"
},
"to_vertex": {
"vertex_id_template": "{col3}",
"vertex_definition_id": "(an identifier for the vertex referenced above)"
},
"edge_id_template": {
"label": "(the edge label to add)",
"template": "{col1}_{col3}"
},
"edge_properties":[
{
"property_name": "(the property to add)",
"property_value_template": "{col4} or text",
"property_value_type": "(data type like String, int, double)"
}
]
}
]
}
]
}
```
The presence of a vertex label implies that the vertex is being created here. Its absence implies that the vertex is created by a different source, and this definition is only adding vertex properties. Specify as many vertex and edge definitions as required to specify the mappings for your entire relational database source.
A sample rule for an `employee` table follows.
```
{
"rules": [
{
"rule_id": "1",
"rule_name": "vertex_mapping_rule_from_nodes",
"table_name": "nodes",
"vertex_definitions": [
{
"vertex_id_template": "{emp_id}",
"vertex_label": "employee",
"vertex_definition_id": "1",
"vertex_properties": [
{
"property_name": "name",
"property_value_template": "{emp_name}",
"property_value_type": "String"
}
]
}
]
},
{
"rule_id": "2",
"rule_name": "edge_mapping_rule_from_emp",
"table_name": "nodes",
"edge_definitions": [
{
"from_vertex": {
"vertex_id_template": "{emp_id}",
"vertex_definition_id": "1"
},
"to_vertex": {
"vertex_id_template": "{mgr_id}",
"vertex_definition_id": "1"
},
"edge_id_template": {
"label": "reportsTo",
"template": "{emp_id}_{mgr_id}"
},
"edge_properties":[
{
"property_name": "team",
"property_value_template": "{team}",
"property_value_type": "String"
}
]
}
]
}
]
}
```
Here, the vertex and edge definitions map a reporting relationship from an `employee` node with employee ID (`EmpID`) and an `employee` node with a manager ID (`managerId`).
For more information about creating graph-mapping rules using Gremlin JSON, see [Gremlin load data format](https://docs.aws.amazon.com/neptune/latest/userguide/bulk-load-tutorial-format-gremlin.html) in the *Amazon Neptune User Guide*.
### Graph-mapping rules for generating RDF/SPARQL data
If you are loading RDF data to be queried using SPARQL, write the graph-mapping rules in R2RML. R2RML is a standard W3C language for mapping relational data to RDF. In an R2RML file, a *triples map* (for example, `<#TriplesMap1>` following) specifies a rule for translating each row of a logical table to zero or more RDF triples. A *subject map* (for example, any `rr:subjectMap` following) specifies a rule for generating the subjects of the RDF triples generated by a triples map. A *predicate-object map* (for example, any `rr:predicateObjectMap` following) is a function that creates one or more predicate-object pairs for each logical table row of a logical table.
A simple example for a `nodes` table follows.
```
@prefix rr: .
@prefix ex: .
<#TriplesMap1>
rr:logicalTable [ rr:tableName "nodes" ];
rr:subjectMap [
rr:template "http://data.example.com/employee/{id}";
rr:class ex:Employee;
];
rr:predicateObjectMap [
rr:predicate ex:name;
rr:objectMap [ rr:column "label" ];
]
```
In the previous example, the mapping defines graph nodes mapped from a table of employees.
Another simple example for a `Student` table follows.
```
@prefix rr: .
@prefix ex: .
@prefix foaf: .
@prefix xsd: .
<#TriplesMap2>
rr:logicalTable [ rr:tableName "Student" ];
rr:subjectMap [ rr:template "http://example.com/{ID}{Name}";
rr:class foaf:Person ];
rr:predicateObjectMap [
rr:predicate ex:id ;
rr:objectMap [ rr:column "ID";
rr:datatype xsd:integer ]
];
rr:predicateObjectMap [
rr:predicate foaf:name ;
rr:objectMap [ rr:column "Name" ]
].
```
In the previous example, the mapping defines graph nodes mapping friend-of-a-friend relationships between persons in a `Student` table.
For more information about creating graph-mapping rules using SPARQL R2RML, see the W3C specification [R2RML: RDB to RDF mapping language](https://www.w3.org/TR/r2rml/).
## Data types for Gremlin and R2RML migration to Amazon Neptune as a target
AWS DMS performs data type mapping from your SQL source endpoint to your Neptune target in one of two ways. Which way you use depends on the graph mapping format that you're using to load the Neptune database:
+ Apache TinkerPop Gremlin, using a JSON representation of the migration data.
+ W3C's SPARQL, using an R2RML representation of the migration data.
For more information on these two graph mapping formats, see [Specifying graph-mapping rules using Gremlin and R2RML for Amazon Neptune as a target](#CHAP_Target.Neptune.GraphMapping).
Following, you can find descriptions of the data type mappings for each format.
### SQL source to Gremlin target data type mappings
The following table shows the data type mappings from a SQL source to a Gremlin formatted target.
AWS DMS maps any unlisted SQL source data type to a Gremlin `String`.
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Neptune.html)
For more information on the Gremlin data types for loading Neptune, see [Gremlin data types](https://docs.aws.amazon.com//neptune/latest/userguide/bulk-load-tutorial-format-gremlin.html#bulk-load-tutorial-format-gremlin-datatypes) in the *Neptune User Guide.*
### SQL source to R2RML (RDF) target data type mappings
The following table shows the data type mappings from a SQL source to an R2RML formatted target.
All listed RDF data types are case-sensitive, except RDF literal. AWS DMS maps any unlisted SQL source data type to an RDF literal.
An *RDF literal* is one of a variety of literal lexical forms and data types. For more information, see [RDF literals](https://www.w3.org/TR/2004/REC-rdf-concepts-20040210/#section-Graph-Literal) in the W3C specification *Resource Description Framework (RDF): Concepts and Abstract Syntax*.
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Neptune.html)
For more information on the RDF data types for loading Neptune and their mappings to SQL source data types, see [Datatype conversions](https://www.w3.org/TR/r2rml/#datatype-conversions) in the W3C specification *R2RML: RDB to RDF Mapping Language*.
## Limitations of using Amazon Neptune as a target
The following limitations apply when using Neptune as a target:
+ AWS DMS currently supports full load tasks only for migration to a Neptune target. Change data capture (CDC) migration to a Neptune target isn't supported.
+ Make sure that your target Neptune database is manually cleared of all data before starting the migration task, as in the following examples.
To drop all data (vertices and edges) within the graph, run the following Gremlin command.
```
gremlin> g.V().drop().iterate()
```
To drop vertices that have the label `'customer'`, run the following Gremlin command.
```
gremlin> g.V().hasLabel('customer').drop()
```
**Note**
It can take some time to drop a large dataset. You might want to iterate `drop()` with a limit, for example, `limit(1000)`.
To drop edges that have the label `'rated'`, run the following Gremlin command.
```
gremlin> g.E().hasLabel('rated').drop()
```
**Note**
It can take some time to drop a large dataset. You might want to iterate `drop()` with a limit, for example `limit(1000)`.
+ The DMS API operation `DescribeTableStatistics` can return inaccurate results about a given table because of the nature of Neptune graph data structures.
During migration, AWS DMS scans each source table and uses graph mapping to convert the source data into a Neptune graph. The converted data is first stored in the S3 bucket folder specified for the target endpoint. If the source is scanned and this intermediate S3 data is generated successfully, `DescribeTableStatistics` assumes that the data was successfully loaded into the Neptune target database. But this isn't always true. To verify that the data was loaded correctly for a given table, compare `count()` return values at both ends of the migration for that table.
In the following example, AWS DMS has loaded a `customer` table from the source database, which is assigned the label `'customer'` in the target Neptune database graph. You can make sure that this label is written to the target database. To do this, compare the number of `customer` rows available from the source database with the number of `'customer'` labeled rows loaded in the Neptune target database after the task completes.
To get the number of customer rows available from the source database using SQL, run the following.
```
select count(*) from customer;
```
To get the number of `'customer'` labeled rows loaded into the target database graph using Gremlin, run the following.
```
gremlin> g.V().hasLabel('customer').count()
```
+ Currently, if any single table fails to load, the whole task fails. Unlike in a relational database target, data in Neptune is highly connected, which makes it impossible in many cases to resume a task. If a task can't be resumed successfully because of this type of data load failure, create a new task to load the table that failed to load. Before running this new task, manually clear the partially loaded table from the Neptune target.
**Note**
You can resume a task that fails migration to a Neptune target if the failure is recoverable (for example, a network transit error).
+ AWS DMS supports most standards for R2RML. However, AWS DMS doesn't support certain R2RML standards, including inverse expressions, joins, and views. A work-around for an R2RML view is to create a corresponding custom SQL view in the source database. In the migration task, use table mapping to choose the view as input. Then map the view to a table that is then consumed by R2RML to generate graph data.
+ When you migrate source data with unsupported SQL data types, the resulting target data can have a loss of precision. For more information, see [Data types for Gremlin and R2RML migration to Amazon Neptune as a target](#CHAP_Target.Neptune.DataTypes).
+ AWS DMS doesn't support migrating LOB data into a Neptune target.
# Using Redis OSS as a target for AWS Database Migration Service
Redis OSS is an open-source in-memory data structure store used as a database, cache, and message broker. Managing data in-memory can result in read or write operations taking less than a millisecond, and hundreds of millions of operations performed each second. As an in-memory data store, Redis OSS powers the most demanding applications requiring sub-millisecond response times.
Using AWS DMS, you can migrate data from any supported source database to a target Redis OSS data store with minimal downtime. For additional information about Redis OSS see, [Redis OSS Documentation](https://redis.io/documentation).
In addition to on-premises Redis OSS, AWS Database Migration Service supports the following:
+ [Amazon ElastiCache (Redis OSS)](https://aws.amazon.com/elasticache/redis/) as a target data store. ElastiCache (Redis OSS) works with your Redis OSS clients and uses the open Redis OSS data format to store your data.
+ [Amazon MemoryDB](https://aws.amazon.com/memorydb/) as a target data store. MemoryDB is compatible with Redis OSS and enables you to build applications using all the Redis OSS data structures, APIs, and commands in use today.
For additional information about working with Redis OSS as a target for AWS DMS, see the following sections:
**Topics**
+ [Prerequisites for using a Redis OSS cluster as a target for AWS DMS](#CHAP_Target.Redis.Prerequisites)
+ [Limitations when using Redis as a target for AWS Database Migration Service](#CHAP_Target.Redis.Limitations)
+ [Migrating data from a relational or non-relational database to a Redis OSS target](#CHAP_Target.Redis.Migrating)
+ [Specifying endpoint settings for Redis OSS as a target](#CHAP_Target.Redis.EndpointSettings)
## Prerequisites for using a Redis OSS cluster as a target for AWS DMS
DMS supports an on-premises Redis OSS target in a standalone configuration, or as a Redis OSS cluster where data is automatically *sharded* across multiple nodes. Sharding is the process of separating data into smaller chunks called shards that are spread across multiple servers or nodes. In effect, a shard is a data partition that contains a subset of the total data set, and serves a slice of the overall workload.
Since Redis OSS is a key-value NoSQL data store, the Redis OSS key naming convention to use when your source is a relational database, is **schema-name.table-name.primary-key**. In Redis OSS, the key and value must not contain the special character %. Otherwise, DMS skips the record.
**Note**
If you are using ElastiCache (Redis OSS) as a target, DMS supports *cluster mode enabled* configurations only. For more information about using ElastiCache (Redis OSS) version 6.x or higher to create a cluster mode enabled target data store, see [Getting started](https://docs.aws.amazon.com/AmazonElastiCache/latest/red-ug/GettingStarted.html) in the *Amazon ElastiCache (Redis OSS) User Guide*.
Before you begin a database migration, launch your Redis OSS cluster with the following criteria.
+ Your cluster has one or more shards.
+ If you're using an ElastiCache (Redis OSS) target, ensure that your cluster doesn't use IAM role-based access control. Instead, use Redis OSS Auth to authenticate users.
+ Enable Multi-AZ (Availability Zones).
+ Ensure the cluster has sufficient memory available to fit the data to be migrated from your database.
+ Make sure that your target Redis OSS cluster is clear of all data before starting the initial migration task.
You should determine your security requirements for the data migration prior to creating your cluster configuration. DMS supports migration to target replication groups regardless of their encryption configuration. But you can enable or disable encryption only when you create your cluster configuration.
## Limitations when using Redis as a target for AWS Database Migration Service
The following limitations apply when using Redis OSS as a target:
+ Since Redis OSS is a key-value no-sql data store, the Redis OSS key naming convention to use when your source is a relational database, is `schema-name.table-name.primary-key`.
+ In Redis OSS, the key-value can't contain the special character `%`. Otherwise, DMS skips the record.
+ DMS won't migrate rows that contain the `%` character.
+ DMS won't migrate fields that contain the `%` character in the field name.
+ Full LOB mode is not supported.
+ A private Certificate Authority (CA) isn’t supported when using ElastiCache (Redis OSS) as a target.
+ AWS DMS does not support source data containing embedded `'\0'` characters when using Redis as a target endpoint. Data containing embedded `'\0'` characters will be truncated at the first `'\0'` character.
## Migrating data from a relational or non-relational database to a Redis OSS target
You can migrate data from any source SQL or NoSQL data store directly to a Redis OSS target. Setting up and starting a migration to a Redis OSS target is similar to any full load and change data capture migration using the DMS console or API. To perform a database migration to a Redis OSS target, you do the following.
+ Create a replication instance to perform all the processes for the migration. For more information, see [Creating a replication instance](CHAP_ReplicationInstance.Creating.md).
+ Specify a source endpoint. For more information, see [Creating source and target endpoints](CHAP_Endpoints.Creating.md).
+ Locate the DNS name and port number of your cluster.
+ Download a certificate bundle that you can use to verify SSL connections.
+ Specify a target endpoint, as described below.
+ Create a task or set of tasks to define what tables and replication processes you want to use. For more information, see [Creating a task](CHAP_Tasks.Creating.md).
+ Migrate data from your source database to your target cluster.
You begin a database migration in one of two ways:
1. You can choose the AWS DMS console and perform each step there.
1. You can use the AWS Command Line Interface (AWS CLI). For more information about using the CLI with AWS DMS, see [AWS CLI for AWS DMS](http://docs.aws.amazon.com/cli/latest/reference/dms/index.html).
**To locate the DNS name and port number of your cluster**
+ Use the following AWS CLI command to provide the `replication-group-id` with the name of your replication group.
```
aws elasticache describe-replication-groups --replication-group-id myreplgroup
```
Here, the output shows the DNS name in the `Address` attribute and the port number in the `Port` attribute of the primary node in the cluster.
```
...
"ReadEndpoint": {
"Port": 6379,
"Address": "myreplgroup-
111.1abc1d.1111.uuu1.cache.example.com"
}
...
```
If you are using MemoryDB as your target, use the following AWS CLI command to provide an endpoint address to your Redis OSS cluster.
```
aws memorydb describe-clusters --clusterid clusterid
```
**Download a certificate bundle for use to verify SSL connections**
+ Enter the following `wget` command at the command line. Wget is a free GNU command-line utility tool used to download files from the internet.
```
wget https://s3.aws-api-domain/rds-downloads/rds-combined-ca-bundle.pem
```
Here, `aws-api-domain` completes the Amazon S3 domain in your AWS Region required to access the specified S3 bucket and the rds-combined-ca-bundle.pem file that it provides.
**To create a target endpoint using the AWS DMS console**
This endpoint is for your Redis OSS target that is already running.
+ On the console, choose **Endpoints** from the navigation pane and then choose **Create Endpoint**. The following table describes the settings.
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Redis.html)
When you're finished providing all information for your endpoint, AWS DMS creates your Redis OSS target endpoint for use during database migration.
For information about creating a migration task and starting your database migration, see [Creating a task](CHAP_Tasks.Creating.md).
## Specifying endpoint settings for Redis OSS as a target
To create or modify a target endpoint, you can use the console or the `CreateEndpoint` or `ModifyEndpoint` API operations.
For a Redis OSS target in the AWS DMS console, specify **Endpoint-specific settings** on the **Create endpoint** or **Modify endpoint** console page.
When using `CreateEndpoint` and `ModifyEndpoint` API operations, specify request parameters for the `RedisSettings` option. The example following shows how to do this using the AWS CLI.
```
aws dms create-endpoint --endpoint-identifier my-redis-target
--endpoint-type target --engine-name redis --redis-settings
'{"ServerName":"sample-test-sample.zz012zz.cluster.eee1.cache.bbbxxx.com","Port":6379,"AuthType":"auth-token",
"SslSecurityProtocol":"ssl-encryption", "AuthPassword":"notanactualpassword"}'
{
"Endpoint": {
"EndpointIdentifier": "my-redis-target",
"EndpointType": "TARGET",
"EngineName": "redis",
"EngineDisplayName": "Redis",
"TransferFiles": false,
"ReceiveTransferredFiles": false,
"Status": "active",
"KmsKeyId": "arn:aws:kms:us-east-1:999999999999:key/x-b188188x",
"EndpointArn": "arn:aws:dms:us-east-1:555555555555:endpoint:ABCDEFGHIJKLMONOPQRSTUVWXYZ",
"SslMode": "none",
"RedisSettings": {
"ServerName": "sample-test-sample.zz012zz.cluster.eee1.cache.bbbxxx.com",
"Port": 6379,
"SslSecurityProtocol": "ssl-encryption",
"AuthType": "auth-token"
}
}
}
```
The `--redis-settings` parameters follow:
+ `ServerName`–(Required) Of type `string`, specifies the Redis OSS cluster that data will be migrated to, and is in your same VPC.
+ `Port`–(Required) Of type `number`, the port value used to access the endpoint.
+ `SslSecurityProtocol`–(Optional) Valid values include `plaintext` and `ssl-encryption`. The default is `ssl-encryption`.
The `plaintext` option doesn't provide Transport Layer Security (TLS) encryption for traffic between endpoint and database.
Use `ssl-encryption` to make an encrypted connection. `ssl-encryption` doesn’t require an SSL Certificate Authority (CA) ARN to verify a server’s certificate, but one can be identified optionally using the `SslCaCertificateArn` setting. If a certificate authority ARN isn't given, DMS uses the Amazon root CA.
When using an on-premises Redis OSS target, you can use `SslCaCertificateArn` to import public or private Certificate Authority (CA) into DMS, and provide that ARN for server authentication. A private CA isn’t supported when using ElastiCache (Redis OSS) as a target.
+ `AuthType`–(Required) Indicates the type of authentication to perform when connecting to Redis OSS. Valid values include `none`, `auth-token`, and `auth-role`.
The `auth-token` option requires an "*AuthPassword*" be provided, while the `auth-role` option requires "*AuthUserName*" and "*AuthPassword*" be provided.
# Using Babelfish as a target for AWS Database Migration Service
You can migrate data from a Microsoft SQL Server source database to a Babelfish target using AWS Database Migration Service.
Babelfish for Aurora PostgreSQL extends your Amazon Aurora PostgreSQL-Compatible Edition database with the ability to accept database connections from Microsoft SQL Server clients. Doing this allows applications originally built for SQL Server to work directly with Aurora PostgreSQL with few code changes compared to a traditional migration, and without changing database drivers.
For information about versions of Babelfish that AWS DMS supports as a target, see [Targets for AWS DMS](CHAP_Introduction.Targets.md). Earlier versions of Babelfish on Aurora PostgreSQL require an upgrade before using the Babelfish endpoint.
**Note**
The Aurora PostgreSQL target endpoint is the preferred way to migrate data to Babelfish. For more information, see [Using Babelfish for Aurora PostgreSQL as a target](CHAP_Target.PostgreSQL.md#CHAP_Target.PostgreSQL.Babelfish).
For information about using Babelfish as a database endpoint, see [Babelfish for Aurora PostgreSQL](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Aurora.AuroraPostgreSQL.html) in the *Amazon Aurora User Guide for Aurora*
## Prerequisites to using Babelfish as a target for AWS DMS
You must create your tables before migrating data to make sure that AWS DMS uses the correct data types and table metadata. If you don't create your tables on the target before running migration, AWS DMS may create the tables with incorrect data types and permissions. For example, AWS DMS creates a timestamp column as binary(8) instead, and doesn't provide the expected timestamp/rowversion functionality.
**To prepare and create your tables prior to migration**
1. Run your create table DDL statements that include any unique constraints, primary keys, or default constraints.
Do not include foreign key constraints, or any DDL statements for objects like views, stored procedures, functions, or triggers. You can apply them after migrating your source database.
1. Identify any identity columns, computed columns, or columns containing rowversion or timestamp data types for your tables. Then, create the necessary transformation rules to handle known issues when running the migration task. For more information see, [Transformation rules and actions](CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Transformations.md).
1. Identify columns with data types that Babelfish doesn't support. Then, change the affected columns in the target table to use supported data types, or create a transformation rule that removes them during the migration task. For more information see, [Transformation rules and actions](CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Transformations.md).
The following table lists source data types not supported by Babelfish, and the corresponding recommended target data type to use.
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Babelfish.html)
**To set Aurora capacity units (ACUs) level for your Aurora PostgreSQL Serverless V2 source database**
You can improve performance of your AWS DMS migration task prior to running it by setting the minimum ACU value.
+ From the **Severless v2 capacity settings** window, set **Minimum ACUs** to **2**, or a reasonable level for your Aurora DB cluster.
For additional information about setting Aurora capacity units, see [ Choosing the Aurora Serverless v2 capacity range for an Aurora cluster](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/aurora-serverless-v2.setting-capacity.html) in the *Amazon Aurora User Guide*
After running your AWS DMS migration task, you can reset the minimum value of your ACUs to a reasonable level for your Aurora PostgreSQL Serverless V2 source database.
## Security requirements when using Babelfish as a target for AWS Database Migration Service
The following describes the security requirements for using AWS DMS with a Babelfish target:
+ The administrator user name (the Admin user) used to create the database.
+ PSQL login and user with the sufficient SELECT, INSERT, UPDATE, DELETE, and REFERENCES permissions.
## User permissions for using Babelfish as a target for AWS DMS
**Important**
For security purposes, the user account used for the data migration must be a registered user in any Babelfish database that you use as a target.
Your Babelfish target endpoint requires minimum user permissions to run an AWS DMS migration.
**To create a login and a low-privileged Transact-SQL (T-SQL) user**
1. Create a login and password to use when connecting to the server.
```
CREATE LOGIN dms_user WITH PASSWORD = 'password';
GO
```
1. Create the virtual database for your Babelfish cluster.
```
CREATE DATABASE my_database;
GO
```
1. Create the T-SQL user for your target database.
```
USE my_database
GO
CREATE USER dms_user FOR LOGIN dms_user;
GO
```
1. For each table in your Babelfish database, GRANT permissions to the tables.
```
GRANT SELECT, DELETE, INSERT, REFERENCES, UPDATE ON [dbo].[Categories] TO dms_user;
```
## Limitations on using Babelfish as a target for AWS Database Migration Service
The following limitations apply when using a Babelfish database as a target for AWS DMS:
+ Only table preparation mode “**Do Nothing**“ is supported.
+ The ROWVERSION data type requires a table mapping rule that removes the column name from the table during the migration task.
+ The sql\$1variant data type isn't supported.
+ Full LOB mode is supported. Using SQL Server as a source endpoint requires the SQL Server Endpoint Connection Attribute setting `ForceFullLob=True` to be set in order for LOBs to be migrated to the target endpoint.
+ Replication task settings have the following limitations:
```
{
"FullLoadSettings": {
"TargetTablePrepMode": "DO_NOTHING",
"CreatePkAfterFullLoad": false,
}.
}
```
+ TIME(7), DATETIME2(7), and DATETIMEOFFSET(7) data types in Babelfish limit the precision value for the seconds portion of the time to 6 digits. Consider using a precision value of 6 for your target table when using these data types. For Babelfish versions 2.2.0 and higher, when using TIME(7) and DATETIME2(7), the seventh digit of precision is always zero.
+ In DO\$1NOTHING mode, DMS checks to see if the table already exists. If the table doesn't exist in the target schema, DMS creates the table based on the source table definition, and maps any user defined data types to their base data type.
+ An AWS DMS migration task to a Babelfish target doesn't support tables that have columns using ROWVERSION or TIMESTAMP data types. You can use a table mapping rule that removes the column name from the table during the transfer process. In the following transformation rule example, a table named `Actor` in your source is transformed to remove all columns starting with the characters `col` from the `Actor` table in your target.
```
{
"rules": [{
"rule-type": "selection",is
"rule-id": "1",
"rule-name": "1",
"object-locator": {
"schema-name": "test",
"table-name": "%"
},
"rule-action": "include"
}, {
"rule-type": "transformation",
"rule-id": "2",
"rule-name": "2",
"rule-action": "remove-column",
"rule-target": "column",
"object-locator": {
"schema-name": "test",
"table-name": "Actor",
"column-name": "col%"
}
}]
}
```
+ For tables with identity or computed columns, where the target tables use mixed case names like Categories, you must create a transformation rule action that converts the table names to lowercase for your DMS task. The following example shows how to create the transformation rule action, **Make lowercase** using the AWS DMS console. For more information, see [Transformation rules and actions](CHAP_Tasks.CustomizingTasks.TableMapping.SelectionTransformation.Transformations.md).
![\[Babelfish transformation rule\]](http://docs.aws.amazon.com/dms/latest/userguide/images/datarep-babelfish-transform-1.png)
+ Prior to Babelfish version 2.2.0, DMS limited the number of columns that you could replicate to a Babelfish target endpoint to twenty (20) columns. With Babelfish 2.2.0 the limit increased to 100 columns. But with Babelfish versions 2.4.0 and higher, the number of columns that you can replicate increases again. You can run the following code sample against your SQL Server database to determine which tables are too long.
```
USE myDB;
GO
DECLARE @Babelfish_version_string_limit INT = 8000; -- Use 380 for Babelfish versions before 2.2.0
WITH bfendpoint
AS (
SELECT
[TABLE_SCHEMA]
,[TABLE_NAME]
, COUNT( [COLUMN_NAME] ) AS NumberColumns
, ( SUM( LEN( [COLUMN_NAME] ) + 3)
+ SUM( LEN( FORMAT(ORDINAL_POSITION, 'N0') ) + 3 )
+ LEN( TABLE_SCHEMA ) + 3
+ 12 -- INSERT INTO string
+ 12) AS InsertIntoCommandLength -- values string
, CASE WHEN ( SUM( LEN( [COLUMN_NAME] ) + 3)
+ SUM( LEN( FORMAT(ORDINAL_POSITION, 'N0') ) + 3 )
+ LEN( TABLE_SCHEMA ) + 3
+ 12 -- INSERT INTO string
+ 12) -- values string
>= @Babelfish_version_string_limit
THEN 1
ELSE 0
END AS IsTooLong
FROM [INFORMATION_SCHEMA].[COLUMNS]
GROUP BY [TABLE_SCHEMA], [TABLE_NAME]
)
SELECT *
FROM bfendpoint
WHERE IsTooLong = 1
ORDER BY TABLE_SCHEMA, InsertIntoCommandLength DESC, TABLE_NAME
;
```
## Target data types for Babelfish
The following table shows the Babelfish target data types that are supported when using AWS DMS and the default mapping from AWS DMS data types.
For additional information about AWS DMS data types, see [Data types for AWS Database Migration Service](CHAP_Reference.DataTypes.md).
| AWS DMS data type | Babelfish data type |
| --- | --- |
| BOOLEAN | TINYINT |
| BYTES | VARBINARY(length) |
| DATE | DATE |
| TIME | TIME |
| INT1 | SMALLINT |
| INT2 | SMALLINT |
| INT4 | INT |
| INT8 | BIGINT |
| NUMERIC | NUMERIC(p,s) |
| REAL4 | REAL |
| REAL8 | FLOAT |
| STRING | If the column is a date or time column, then do the following: [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Babelfish.html) If the column is not a date or time column, use VARCHAR (length). |
| UINT1 | TINYINT |
| UINT2 | SMALLINT |
| UINT4 | INT |
| UINT8 | BIGINT |
| WSTRING | NVARCHAR(length) |
| BLOB | VARBINARY(max) To use this data type with DMS, you must enable the use of BLOBs for a specific task. DMS supports BLOB data types only in tables that include a primary key. |
| CLOB | VARCHAR(max) To use this data type with DMS, you must enable the use of CLOBs for a specific task. |
| NCLOB | NVARCHAR(max) To use this data type with DMS, you must enable the use of NCLOBs for a specific task. During CDC, DMS supports NCLOB data types only in tables that include a primary key. |
# Using Amazon Timestream as a target for AWS Database Migration Service
You can use AWS Database Migration Service to migrate data from your source database to a Amazon Timestream target endpoint, with support for Full Load and CDC data migrations.
Amazon Timestream is a fast, scalable, and serverless time series database service built for high-volume data ingestion. Time series data is a sequence of data points collected over a time interval, and is used for measuring events that change over time. It is used to collect, store, and analyze metrics from IoT applications, DevOps applications, and analytics applications. Once you have your data in Timestream, you can visualize and identify trends and patterns in your data in near real-time. For information about Amazon Timestream, see [What is Amazon Timestream?](https://docs.aws.amazon.com/timestream/latest/developerguide/what-is-timestream.html) in the *Amazon Timestream Developer Guide*.
**Topics**
+ [Prerequisites for using Amazon Timestream as a target for AWS Database Migration Service](#CHAP_Target.Timestream.Prerequisites)
+ [Multithreaded full load task settings](#CHAP_Target.Timestream.FLTaskSettings)
+ [Multithreaded CDC load task settings](#CHAP_Target.Timestream.CDCTaskSettings)
+ [Endpoint settings when using Timestream as a target for AWS DMS](#CHAP_Target.Timestream.ConnectionAttrib)
+ [Creating and modifying an Amazon Timestream target endpoint](#CHAP_Target.Timestream.CreateModifyEndpoint)
+ [Using object mapping to migrate data to a Timestream topic](#CHAP_Target.Timestream.ObjectMapping)
+ [Limitations when using Amazon Timestream as a target for AWS Database Migration Service](#CHAP_Target.Timestream.Limitations)
## Prerequisites for using Amazon Timestream as a target for AWS Database Migration Service
Before you set up Amazon Timestream as a target for AWS DMS, make sure that you create an IAM role. This role must allow AWS DMS to gain access to the data being migrated into Amazon Timestream. The minimum set of access permissions for the role that you use to migrate to Timestream is shown in the following IAM policy.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "AllowDescribeEndpoints",
"Effect": "Allow",
"Action": [
"timestream:DescribeEndpoints"
],
"Resource": "*"
},
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"timestream:ListTables",
"timestream:DescribeDatabase"
],
"Resource": "arn:aws:timestream:us-east-1:123456789012:database/DATABASE_NAME"
},
{
"Sid": "VisualEditor1",
"Effect": "Allow",
"Action": [
"timestream:DeleteTable",
"timestream:WriteRecords",
"timestream:UpdateTable",
"timestream:CreateTable"
],
"Resource": "arn:aws:timestream:us-east-1:123456789012:database/DATABASE_NAME/table/TABLE_NAME"
}
]
}
```
------
If you intend to migrate all tables, use `*` for *TABLE\$1NAME* in the example above.
Note the following about using Timestream as a target:
+ If you intend to ingest historical data with timestamps exceeding 1 year old, we recommend to use AWS DMS to write the data to Amazon S3 in a comma separated value (csv) format. Then, use Timestream’s batch load to ingest the data into Timestream. For more information, see [Using batch load in Timestream](https://docs.aws.amazon.com/timestream/latest/developerguide/batch-load.html) in the [Amazon Timestream developer guide](https://docs.aws.amazon.com/timestream/latest/developerguide/what-is-timestream.html).
+ For full-load data migrations of data less than 1 year old, we recommend setting the memory store retention period of the Timestream table greater than or equal to the oldest timestamp. Then, once migration completes, edit the table's memory store retention to the desired value. For example, to migrate data with the oldest timestamp being 2 months old, do the following:
+ Set the Timestream target table's memory store retention to 2 months.
+ Start the data migration using AWS DMS.
+ Once the data migration completes, change the retention period of the target Timestream table to your desired value.
We recommend estimating the memory store cost prior to the migration, using the information on the following pages:
+ [Amazon Timestream pricing](https://aws.amazon.com/timestream/pricing)
+ [AWS pricing calculator](https://calculator.aws/#/addService)
+ For CDC data migrations, we recommend setting the memory store retention period of the target table such that ingested data falls within the memory store retention bounds. For more information, see [ Writes Best Practices ](https://docs.aws.amazon.com/timestream/latest/developerguide/data-ingest.html) in the [Amazon Timestream developer guide](https://docs.aws.amazon.com/timestream/latest/developerguide/what-is-timestream.html).
## Multithreaded full load task settings
To help increase the speed of data transfer, AWS DMS supports a multithreaded full load migration task to a Timestream target endpoint with these task settings:
+ `MaxFullLoadSubTasks` – Use this option to indicate the maximum number of source tables to load in parallel. DMS loads each table into its corresponding Amazon Timestream target table using a dedicated subtask. The default is 8; the maximum value is 49.
+ `ParallelLoadThreads` – Use this option to specify the number of threads that AWS DMS uses to load each table into its Amazon Timestream target table. The maximum value for a Timestream target is 32. You can ask to have this maximum limit increased.
+ `ParallelLoadBufferSize` – Use this option to specify the maximum number of records to store in the buffer that the parallel load threads use to load data to the Amazon Timestream target. The default value is 50. The maximum value is 1,000. Use this setting with `ParallelLoadThreads`. `ParallelLoadBufferSize` is valid only when there is more than one thread.
+ `ParallelLoadQueuesPerThread` – Use this option to specify the number of queues each concurrent thread accesses to take data records out of queues and generate a batch load for the target. The default is 1. However, for Amazon Timestream targets of various payload sizes, the valid range is 5–512 queues per thread.
## Multithreaded CDC load task settings
To promote CDC performance, AWS DMS supports these task settings:
+ `ParallelApplyThreads` – Specifies the number of concurrent threads that AWS DMS uses during a CDC load to push data records to a Timestream target endpoint. The default value is 0 and the maximum value is 32.
+ `ParallelApplyBufferSize` – Specifies the maximum number of records to store in each buffer queue for concurrent threads to push to a Timestream target endpoint during a CDC load. The default value is 100 and the maximum value is 1,000. Use this option when `ParallelApplyThreads` specifies more than one thread.
+ `ParallelApplyQueuesPerThread` – Specifies the number of queues that each thread accesses to take data records out of queues and generate a batch load for a Timestream endpoint during CDC. The default value is 1 and the maximum value is 512.
## Endpoint settings when using Timestream as a target for AWS DMS
You can use endpoint settings to configure your Timestream target database similar to using extra connection attributes. You specify the settings when you create the target endpoint using the AWS DMS console, or by using the `create-endpoint` command in the [AWS CLI](https://docs.aws.amazon.com/cli/latest/reference/dms/index.html), with the `--timestream-settings '{"EndpointSetting": "value", ...}'` JSON syntax.
The following table shows the endpoint settings that you can use with Timestream as a target.
[\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Timestream.html)
## Creating and modifying an Amazon Timestream target endpoint
Once you have created an IAM role and established the minimum set of access permissions, you can create a Amazon Timestream target endpoint using the AWS DMS console, or by using the `create-endpoint` command in the [AWS CLI](https://docs.aws.amazon.com/cli/latest/reference/dms/index.html), with the `--timestream-settings '{"EndpointSetting": "value", ...}'` JSON syntax.
The following examples show how to create and modify a Timestream target endpoint using the AWS CLI.
**Create Timestream target endpoint command**
```
aws dms create-endpoint —endpoint-identifier timestream-target-demo
--endpoint-type target —engine-name timestream
--service-access-role-arn arn:aws:iam::123456789012:role/my-role
--timestream-settings
{
"MemoryDuration": 20,
"DatabaseName":"db_name",
"MagneticDuration": 3,
"CdcInsertsAndUpdates": true,
"EnableMagneticStoreWrites": true,
}
```
**Modify Timestream target endpoint command**
```
aws dms modify-endpoint —endpoint-identifier timestream-target-demo
--endpoint-type target —engine-name timestream
--service-access-role-arn arn:aws:iam::123456789012:role/my-role
--timestream-settings
{
"MemoryDuration": 20,
"MagneticDuration": 3,
}
```
## Using object mapping to migrate data to a Timestream topic
AWS DMS uses table-mapping rules to map data from the source to the target Timestream topic. To map data to a target topic, you use a type of table-mapping rule called object mapping. You use object mapping to define how data records in the source map to the data records published to a Timestream topic.
Timestream topics don't have a preset structure other than having a partition key.
**Note**
You don't have to use object mapping. You can use regular table mapping for various transformations. However, the partition key type will follow these default behaviors:
Primary Key is used as a partition key for Full Load.
If no parallel-apply task settings are used, `schema.table` is used as a partition key for CDC.
If parallel-apply task settings are used, Primary key is used as a partition key for CDC.
To create an object-mapping rule, specify `rule-type` as `object-mapping`. This rule specifies what type of object mapping you want to use. The structure for the rule is as follows.
```
{
"rules": [
{
"rule-type": "object-mapping",
"rule-id": "id",
"rule-name": "name",
"rule-action": "valid object-mapping rule action",
"object-locator": {
"schema-name": "case-sensitive schema name",
"table-name": ""
}
}
]
}
```
```
{
"rules": [
{
"rule-type": "object-mapping",
"rule-id": "1",
"rule-name": "timestream-map",
"rule-action": "map-record-to-record",
"target-table-name": "tablename",
"object-locator": {
"schema-name": "",
"table-name": ""
},
"mapping-parameters": {
"timestream-dimensions": [
"column_name1",
"column_name2"
],
"timestream-timestamp-name": "time_column_name",
"timestream-multi-measure-name": "column_name1or2",
"timestream-hash-measure-name": true or false,
"timestream-memory-duration": x,
"timestream-magnetic-duration": y
}
}
]
}
```
AWS DMS currently supports `map-record-to-record` and `map-record-to-document` as the only valid values for the `rule-action` parameter. The `map-record-to-record` and `map-record-to-document` values specify what AWS DMS does by default to records that aren't excluded as part of the `exclude-columns` attribute list. These values don't affect the attribute mappings in any way.
Use `map-record-to-record` when migrating from a relational database to a Timestream topic. This rule type uses the `taskResourceId.schemaName.tableName` value from the relational database as the partition key in the Timestream topic and creates an attribute for each column in the source database. When using `map-record-to-record`, for any column in the source table not listed in the `exclude-columns` attribute list, AWS DMS creates a corresponding attribute in the target topic. This corresponding attribute is created regardless of whether that source column is used in an attribute mapping.
One way to understand `map-record-to-record` is to see it in action. For this example, assume that you are starting with a relational database table row with the following structure and data.
| FirstName | LastName | StoreId | HomeAddress | HomePhone | WorkAddress | WorkPhone | DateofBirth |
| --- | --- | --- | --- | --- | --- | --- | --- |
| Randy | Marsh | 5 | 221B Baker Street | 1234567890 | 31 Spooner Street, Quahog | 9876543210 | 02/29/1988 |
To migrate this information from a schema named `Test` to a Timestream topic, you create rules to map the data to the target topic. The following rule illustrates the mapping.
```
{
"rules": [
{
"rule-type": "selection",
"rule-id": "1",
"rule-name": "1",
"rule-action": "include",
"object-locator": {
"schema-name": "Test",
"table-name": "%"
}
},
{
"rule-type": "object-mapping",
"rule-id": "2",
"rule-name": "DefaultMapToTimestream",
"rule-action": "map-record-to-record",
"object-locator": {
"schema-name": "Test",
"table-name": "Customers"
}
}
]
}
```
Given a Timestream topic and a partition key (in this case, `taskResourceId.schemaName.tableName`), the following illustrates the resulting record format using our sample data in the Timestream target topic:
```
{
"FirstName": "Randy",
"LastName": "Marsh",
"StoreId": "5",
"HomeAddress": "221B Baker Street",
"HomePhone": "1234567890",
"WorkAddress": "31 Spooner Street, Quahog",
"WorkPhone": "9876543210",
"DateOfBirth": "02/29/1988"
}
```
## Limitations when using Amazon Timestream as a target for AWS Database Migration Service
The following limitations apply when using Amazon Timestream as a target:
+ **Dimensions and Timestamps:** Timestream uses the dimensions and timestamps in the source data like a composite primary key, and also does not allow you to upsert these values. This means that if you change the timestamp or the dimensions for a record in the source database, the Timestream database will try to create a new record. It is thus possible that if you change the dimension or timestamp of a record such that they match those of another existing record, then AWS DMS updates the values of the other record instead of creating a new record or updating the previous corresponding record.
+ **DDL Commands:** The current release of AWS DMS only supports `CREATE TABLE` and `DROP TABLE` DDL commands.
+ **Record Limitations:** Timestream has limitations for records such as record size and measure size. For more information, see [Quotas](https://docs.aws.amazon.com/timestream/latest/developerguide/what-is-timestream.html) in the [Amazon Timestream Developer Guide](https://docs.aws.amazon.com/).
+ **Deleting Records and Null Values: ** Timestream doesn't support deleting records. To support migrating records deleted from the source, AWS DMS clears the corresponding fields in the records in the Timestream target database. AWS DMS changes the values in the fields of the corresponding target record with **0** for numeric fields, **null** for text fields, and **false** for boolean fields.
+ Timestream as a target doesn't support sources that aren't relational databases (RDBMS).
+ AWS DMS only supports Timestream as a target in the following regions:
+ US East (N. Virginia)
+ US East (Ohio)
+ US West (Oregon)
+ Europe (Ireland)
+ Europe (Frankfurt)
+ Asia Pacific (Sydney)
+ Asia Pacific (Tokyo)
+ Timestream as a target doesn't support setting `TargetTablePrepMode` to `TRUNCATE_BEFORE_LOAD`. We recommend using `DROP_AND_CREATE` for this setting.
# Using Amazon RDS for Db2 and IBM Db2 LUW as a target for AWS DMS
You can migrate data to an Amazon RDS for Db2 or an on-premises Db2 database from a Db2 LUW database using AWS Database Migration Service (AWS DMS).
For information about versions of Db2 LUW that AWS DMS supports as a target, see [Targets for AWS DMS](CHAP_Introduction.Targets.md).
You can use Secure Sockets Layer (SSL) to encrypt connections between your Db2 LUW endpoint and the replication instance. For more information about using SSL with a Db2 LUW endpoint, see [Using SSL with AWS Database Migration Service](CHAP_Security.SSL.md).
## Limitations when using Db2 LUW as a target for AWS DMS
The following limitations apply when using Db2 LUW database as a target for AWS DMS. For limitations on using Db2 LUW as a source, see [Limitations when using Db2 LUW as a source for AWS DMS](CHAP_Source.DB2.md#CHAP_Source.DB2.Limitations).
+ AWS DMS only supports Db2 LUW as a target when the source is either Db2 LUW or Db2 for z/OS.
+ Using Db2 LUW as a target doesn't support replications with full LOB mode.
+ Using Db2 LUW as a target doesn't support the XML datatype in the full load phase. This is a limitation of the IBM dbload utility. For more information, see [The dbload utility](https://www.ibm.com/docs/en/informix-servers/14.10?topic=utilities-dbload-utility) in the *IBM Informix Servers* documentation.
+ AWS DMS truncates BLOB fields with values corresponding to the double quote character ("). This is a limitation of the IBM dbload utility.
+ AWS DMS does not support the parallel full load option when migrating to a Db2 LUW target in DMS version 3.5.3. This option is available from DMS version 3.5.4 or later.
## Endpoint settings when using Db2 LUW as a target for AWS DMS
You can use endpoint settings to configure your Db2 LUW target database similar to using extra connection attributes. You specify the settings when you create the target endpoint using the AWS DMS console, or by using the `create-endpoint` command in the [AWS CLI](https://docs.aws.amazon.com/cli/latest/reference/dms/index.html), with the `--ibm-db2-settings '{"EndpointSetting": "value", ...}'` JSON syntax.
The following table shows the endpoint settings that you can use with Db2 LUW as a target.
| Name | Description |
| --- | --- |
| `KeepCsvFiles` | If true, AWS DMS saves any .csv files to the Db2 LUW target that were used to replicate data. DMS uses these files for analysis and troubleshooting.. |
| `LoadTimeout` | The amount of time (in milliseconds) before AWS DMS times out operations performed by DMS on the Db2 target. The default value is 1200 (20 minutes). |
| `MaxFileSize` | Specifies the maximum size (in KB) of .csv files used to transfer data to Db2 LUW. |
| `WriteBufferSize` | The size (in KB) of the in-memory file write buffer used when generating .csv files on the local disk on the DMS replication instance. The default value is 1024 (1 MB). |