AWS Data Pipeline 不再提供給新客戶。的現有客戶 AWS Data Pipeline 可以繼續正常使用服務。[進一步了解](https://aws.amazon.com/blogs/big-data/migrate-workloads-from-aws-data-pipeline/)
本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# RedshiftCopyActivity
將資料從 DynamoDB 或 Amazon S3 複製到 Amazon Redshift。您可以將資料載入新的資料表，或是輕鬆地將資料併入現有資料表。
以下是使用 `RedshiftCopyActivity` 的使用案例概觀：
1. 首先使用 AWS Data Pipeline 在 Amazon S3 中暫存您的資料。
1. 使用 將資料從 Amazon RDS 和 Amazon EMR `RedshiftCopyActivity`移至 Amazon Redshift。
這可讓您將資料載入 Amazon Redshift，以便進行分析。
1. 使用 [SqlActivity](dp-object-sqlactivity.md)對已載入 Amazon Redshift 的資料執行 SQL 查詢。
此外，`RedshiftCopyActivity` 可讓您使用 `S3DataNode`，因為它支援資訊清單檔案。如需詳細資訊，請參閱[S3DataNode](dp-object-s3datanode.md)。
## 範例
以下為此物件類型的範例。
為了確保格式轉換，此範例使用 中的 [EMPTYASNULL](https://docs.aws.amazon.com/redshift/latest/dg/copy-parameters-data-conversion.html#copy-emptyasnull) 和 [IGNOREBLANKLINES](https://docs.aws.amazon.com/redshift/latest/dg/copy-parameters-data-conversion.html#copy-ignoreblanklines) 特殊轉換參數`commandOptions`。如需詳細資訊，請參閱《*Amazon Redshift 資料庫開發人員指南*》中的[資料轉換參數](https://docs.aws.amazon.com/redshift/latest/dg/copy-parameters-data-conversion.html)。
```
{
"id" : "S3ToRedshiftCopyActivity",
"type" : "RedshiftCopyActivity",
"input" : { "ref": "MyS3DataNode" },
"output" : { "ref": "MyRedshiftDataNode" },
"insertMode" : "KEEP_EXISTING",
"schedule" : { "ref": "Hour" },
"runsOn" : { "ref": "MyEc2Resource" },
"commandOptions": ["EMPTYASNULL", "IGNOREBLANKLINES"]
}
```
以下範例管道定義會顯示使用 `APPEND` 插入模式的活動：
```
{
"objects": [
{
"id": "CSVId1",
"name": "DefaultCSV1",
"type": "CSV"
},
{
"id": "RedshiftDatabaseId1",
"databaseName": "dbname",
"username": "user",
"name": "DefaultRedshiftDatabase1",
"*password": "password",
"type": "RedshiftDatabase",
"clusterId": "redshiftclusterId"
},
{
"id": "Default",
"scheduleType": "timeseries",
"failureAndRerunMode": "CASCADE",
"name": "Default",
"role": "DataPipelineDefaultRole",
"resourceRole": "DataPipelineDefaultResourceRole"
},
{
"id": "RedshiftDataNodeId1",
"schedule": {
"ref": "ScheduleId1"
},
"tableName": "orders",
"name": "DefaultRedshiftDataNode1",
"createTableSql": "create table StructuredLogs (requestBeginTime CHAR(30) PRIMARY KEY DISTKEY SORTKEY, requestEndTime CHAR(30), hostname CHAR(100), requestDate varchar(20));",
"type": "RedshiftDataNode",
"database": {
"ref": "RedshiftDatabaseId1"
}
},
{
"id": "Ec2ResourceId1",
"schedule": {
"ref": "ScheduleId1"
},
"securityGroups": "MySecurityGroup",
"name": "DefaultEc2Resource1",
"role": "DataPipelineDefaultRole",
"logUri": "s3://myLogs",
"resourceRole": "DataPipelineDefaultResourceRole",
"type": "Ec2Resource"
},
{
"id": "ScheduleId1",
"startDateTime": "yyyy-mm-ddT00:00:00",
"name": "DefaultSchedule1",
"type": "Schedule",
"period": "period",
"endDateTime": "yyyy-mm-ddT00:00:00"
},
{
"id": "S3DataNodeId1",
"schedule": {
"ref": "ScheduleId1"
},
"filePath": "s3://datapipeline-us-east-1/samples/hive-ads-samples.csv",
"name": "DefaultS3DataNode1",
"dataFormat": {
"ref": "CSVId1"
},
"type": "S3DataNode"
},
{
"id": "RedshiftCopyActivityId1",
"input": {
"ref": "S3DataNodeId1"
},
"schedule": {
"ref": "ScheduleId1"
},
"insertMode": "APPEND",
"name": "DefaultRedshiftCopyActivity1",
"runsOn": {
"ref": "Ec2ResourceId1"
},
"type": "RedshiftCopyActivity",
"output": {
"ref": "RedshiftDataNodeId1"
}
}
]
}
```
`APPEND` 操作會將項目新增到資料表，無論其主索引鍵或排序索引鍵為何。例如，若您有以下資料表，您可以使用相同的 ID 和使用者值附加記錄。
```
ID(PK) USER
1 aaa
2 bbb
```
您可以使用相同的 ID 和使用者值附加記錄：
```
ID(PK) USER
1 aaa
2 bbb
1 aaa
```
**注意**
若 `APPEND` 操作遭到插斷並進行重試，其導致的重新執行管道可能會從開頭附加。這可能會造成進一步的重複，因此建議您留意此行為，特別是在您擁有任何計算資料列數量的邏輯時。
如需教學，請參閱[使用 將資料複製到 Amazon Redshift AWS Data Pipeline](dp-copydata-redshift.md)。
## 語法
****
| 必要欄位 | Description | 槽類型 |
| --- | --- | --- |
| insertMode | 決定 AWS Data Pipeline 如何處理目標資料表中與要載入資料中的資料列重疊的預先存在資料。 有效值為：`KEEP_EXISTING`、`OVERWRITE_EXISTING`、`TRUNCATE` 和 `APPEND`。 `KEEP_EXISTING` 會將新列新增至資料表，並保持任何現有列不變。 `KEEP_EXISTING` 和 ` OVERWRITE_EXISTING` 使用主索引鍵、排序索引鍵和分發索引鍵，以識別出哪些傳入的列與現有列匹配。請參閱《Amazon Redshift *資料庫開發人員指南*》中的[更新和插入新資料](https://docs.aws.amazon.com/redshift/latest/dg/t_updating-inserting-using-staging-tables-.html)。 `TRUNCATE` 會刪除目的地資料表中的所有資料，再寫入新資料。 `APPEND` 會將所有記錄新增至 Redshift 資料表結尾。 `APPEND` 不需要主索引鍵、分發索引鍵或排序索引鍵，因此可能會附加可能重複的項目。 | 列舉 |
****
| 物件呼叫欄位 | Description | 槽類型 |
| --- | --- | --- |
| schedule | 在排程間隔的執行期間會呼叫此物件。 指定其他物件的排程參考，以設定此物件的相依性執行順序。 在大部分的情況下，建議將排程參考放在預設的管道物件，讓所有物件都繼承該排程。例如，您可以指定 `"schedule": {"ref": "DefaultSchedule"}` 在物件上明確設定排程。 如果管道中的主排程包含巢狀排程，請建立具有排程參考的父物件。 如需範例選用排程組態的詳細資訊，請參閱[排程](https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-schedule.html)。 | 參考物件，例如： "schedule":\$1"ref":"myScheduleId"\$1 |
****
| 必要的群組 (下列其中之一為必要) | Description | 槽類型 |
| --- | --- | --- |
| runsOn | 執行活動或命令的可運算資源。例如，Amazon EC2 執行個體或 Amazon EMR 叢集。 | 參考物件，例如 "runsOn":\$1"ref":"myResourceId"\$1 |
| workerGroup | 工作者群組。這是用於路由任務。如果您提供 runsOn 值，且 workerGroup 存在，則會忽略 workerGroup。 | String |
****
| 選用欄位 | Description | 槽類型 |
| --- | --- | --- |
| attemptStatus | 遠端活動最新回報的狀態。 | String |
| attemptTimeout | 遠端工作完成的逾時。如果設定，則系統可能會重試未在設定開始時間內完成的遠端活動。 | Period |
| commandOptions | 接受參數以在`COPY`操作期間傳遞至 Amazon Redshift 資料節點。如需參數的資訊，請參閱《Amazon Redshift *資料庫開發人員指南*》中的 [COPY](https://docs.aws.amazon.com/redshift/latest/dg/r_COPY.html)。 `COPY` 載入資料表時，會嘗試隱含地將來源資料中的字串轉換為目標欄的資料類型。除了自動進行的預設資料轉換之外，如果您收到錯誤或有其他轉換需求，您也可以指定其他轉換參數。如需詳細資訊，請參閱《Amazon Redshift *資料庫開發人員指南*》中的[資料轉換參數](https://docs.aws.amazon.com/redshift/latest/dg/copy-parameters-data-conversion.html)。 如果資料格式與輸入或輸出資料節點相關聯，則會忽略提供的參數。 由於複製操作會先使用 `COPY` 將資料插入臨時資料表中，然後使用 `INSERT` 命令將資料從臨時資料表複製到目的地資料表，因此某些 `COPY` 參數會不適用，例如 `COPY` 命令啟用資料表自動壓縮的功能。如果壓縮為必要，請將欄編碼詳細資訊新增至 `CREATE TABLE` 陳述式。 此外，在某些情況下，當需要從 Amazon Redshift 叢集卸載資料並在 Amazon S3 中建立檔案時， `RedshiftCopyActivity`依賴 Amazon Redshift `UNLOAD`的操作。 若要在複製和卸載期間改善效能，請從 `UNLOAD` 命令指定 `PARALLEL OFF` 參數。如需參數的資訊，請參閱《Amazon Redshift *資料庫開發人員指南*》中的 [UNLOAD](https://docs.aws.amazon.com/redshift/latest/dg/r_UNLOAD.html)。 | String |
| dependsOn | 指定與另一個可執行物件的相依性。 | 參考物件："dependsOn":\$1"ref":"myActivityId"\$1 |
| failureAndRerunMode | 描述相依性故障或重新執行時的消費者節點行為 | 列舉 |
| input | 輸入資料節點。資料來源可以是 Amazon S3、DynamoDB 或 Amazon Redshift。 | 參考物件： "input":\$1"ref":"myDataNodeId"\$1 |
| lateAfterTimeout | 物件必須完成的管道啟動後經過的時間。只有在排程類型未設定為 時，才會觸發它ondemand。 | Period |
| maxActiveInstances | 同時作用中的元件執行個體數目上限。重新執行不計入作用中的執行個體數量。 | Integer |
| maximumRetries | 故障時嘗試重試的次數上限 | Integer |
| onFail | 目前物件發生故障時要執行的動作。 | 參考物件："onFail":\$1"ref":"myActionId"\$1 |
| onLateAction | 某個物件尚未排程或仍未完成時，應該觸發的動作。 | 參考物件： "onLateAction":\$1"ref":"myActionId"\$1 |
| onSuccess | 目前物件成功時要執行的動作。 | 參考物件： "onSuccess":\$1"ref":"myActionId"\$1 |
| output | 輸出資料節點。輸出位置可以是 Amazon S3 或 Amazon Redshift。 | 參考物件： "output":\$1"ref":"myDataNodeId"\$1 |
| parent | 目前物件的父系，其插槽會被繼承。 | 參考物件："parent":\$1"ref":"myBaseObjectId"\$1 |
| pipelineLogUri | 上傳管道日誌的 S3 URI (例如 's3://BucketName/Key/')。 | String |
| precondition | 選擇是否定義先決條件。在所有先決條件滿足前，資料節點不會標示為"READY"。 | 參考物件："precondition":\$1"ref":"myPreconditionId"\$1 |
| 佇列 | 對應至 Amazon Redshift 中的 `query_group `設定，可讓您根據並行活動在佇列中的置放來指派並排定其優先順序。 Amazon Redshift 會將同時連線數限制在 15。如需詳細資訊，請參閱《Amazon RDS *資料庫開發人員指南*》中的[將查詢指派給佇列](https://docs.aws.amazon.com/AmazonRDS/latest/DeveloperGuide/cm-c-executing-queries.html)。 | String |
| reportProgressTimeout | 遠端工作連續呼叫 `reportProgress` 的逾時。 如果設定，則不回報指定時段進度的遠端活動，可能會視為已停滯而重試。 | Period |
| retryDelay | 兩次重試嘗試之間的逾時持續時間。 | Period |
| scheduleType | 允許您指定管道中物件的排程。值為：`cron`、`ondemand` 和 `timeseries`。 `timeseries` 排程表示執行個體會排程在每個間隔的結尾。 `Cron` 排程表示執行個體會排程在每個間隔的開頭。 `ondemand` 排程可讓您在每次啟用時執行一次管道。這表示您不必複製或重新建立管道，然後再執行一次。 若要使用 `ondemand` 管道，請針對每次後續執行呼叫 `ActivatePipeline` 操作。 若您使用 `ondemand` 排程，則必須在預設物件中指定此排程，且其必須是針對管道中物件指定的唯一 `scheduleType`。 | 列舉 |
| transformSql | 用於轉換輸入資料的 `SQL SELECT` 表達式。 在資料表執行名為 `staging` 的 `transformSql` 表達式。 當您從 DynamoDB 或 Amazon S3 複製資料時， AWS Data Pipeline 會建立名為「預備」的資料表，並最初載入其中的資料。此資料表中的資料用於更新目標資料表。 `transformSql` 的輸出結構描述，必須與最終目標表格的結構描述相符。 如果您指定 `transformSql` 選項，則會從指定的 SQL 陳述式建立第二個臨時資料表。然後，第二個臨時資料表中的資料會更新於最終目標資料表。 | String |
****
| 執行時間欄位 | Description | 槽類型 |
| --- | --- | --- |
| @activeInstances | 目前已排程的作用中執行個體物件清單。 | 參考物件："activeInstances":\$1"ref":"myRunnableObjectId"\$1 |
| @actualEndTime | 此物件執行完成的時間。 | DateTime |
| @actualStartTime | 此物件執行開始的時間。 | DateTime |
| cancellationReason | 若此物件已取消，會提供 cancellationReason。 | String |
| @cascadeFailedOn | 物件失敗所在的相依鏈的描述。 | 參考物件： "cascadeFailedOn":\$1"ref":"myRunnableObjectId"\$1 |
| emrStepLog | 只在 EMR 活動嘗試時才可使用的 EMR 步驟日誌 | String |
| errorId | 若此物件失敗，會提供 errorId。 | String |
| errorMessage | 若此物件失敗，會提供 errorMessage。 | String |
| errorStackTrace | 如果此物件失敗，則為錯誤堆疊追蹤。 | String |
| @finishedTime | 此物件完成其執行的時間。 | DateTime |
| hadoopJobLog | 嘗試 EMR 型活動可用的 Hadoop 任務日誌。 | String |
| @healthStatus | 反映已達終止狀態之最後一個物件執行個體成功或失敗的物件運作狀態。 | String |
| @healthStatusFromInstanceId | 已達終止狀態之最後一個執行個體物件的 ID。 | String |
| @healthStatusUpdatedTime | 上次更新運作狀態的時間。 | DateTime |
| hostname | 選取任務嘗試之用戶端的主機名稱。 | String |
| @lastDeactivatedTime | 此物件最後停用的時間。 | DateTime |
| @latestCompletedRunTime | 執行完成最近一次執行的時間。 | DateTime |
| @latestRunTime | 執行排程最近一次執行的時間。 | DateTime |
| @nextRunTime | 下次要排程執行的時間。 | DateTime |
| reportProgressTime | 遠端活動最近報告進度的時間。 | DateTime |
| @scheduledEndTime | 物件的排程結束時間。 | DateTime |
| @scheduledStartTime | 物件的排程開始時間。 | DateTime |
| @status | 此物件的狀態。 | String |
| @version | 建立物件使用的管道版本。 | String |
| @waitingOn | 此物件等待之相依性清單的描述。 | 參考物件： "waitingOn":\$1"ref":"myRunnableObjectId"\$1 |
****
| 系統欄位 | Description | 槽類型 |
| --- | --- | --- |
| @error | 描述格式錯誤物件的錯誤。 | String |
| @pipelineId | 此物件所屬管道的 ID。 | String |
| @sphere | 物件的球體。代表其在生命週期中的位置。例如，元件物件會引發執行個體物件，該物件會執行嘗試物件。 | String |