AWS Data Pipeline 不再向新客户提供。的现有客户 AWS Data Pipeline 可以继续照常使用该服务。[了解详情](https://aws.amazon.com/blogs/big-data/migrate-workloads-from-aws-data-pipeline/)
本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 示例
下面是该对象类型的示例。
**Topics**
+ [使用 hadoopVersion 启动 Amazon EMR 集群](emrcluster-example-launch.md)
+ [启动具有版本标签 emr-4.x 或更高版本的 Amazon EMR 集群](emrcluster-example-release-label.md)
+ [在您的 Amazon EMR 集群上安装额外的软件](emrcluster-example-install-software.md)
+ [在 3.x 版本上禁用服务器端加密](emrcluster-example1-disable-encryption.md)
+ [在 4.x 版本上禁用服务器端加密](emrcluster-example2-disable-encryption.md)
+ [配置 Hadoop KMS ACLs 并在 HDFS 中创建加密区域](emrcluster-example-hadoop-kms.md)
+ [指定自定义 IAM 角色](emrcluster-example-custom-iam-roles.md)
+ [使用适用于 Java 的 AWS 开发工具包中的 EmrCluster 资源](emrcluster-example-java.md)
+ [在私有子网中配置 Amazon EMR 集群](emrcluster-example-private-subnet.md)
+ [将 EBS 卷附加到集群节点](emrcluster-example-ebs.md)
# 使用 hadoopVersion 启动 Amazon EMR 集群
**Example**
以下示例使用 AMI 1.0 版和 Hadoop 0.20 启动 Amazon EMR 集群。
```
{
"id" : "MyEmrCluster",
"type" : "EmrCluster",
"hadoopVersion" : "0.20",
"keyPair" : "my-key-pair",
"masterInstanceType" : "m3.xlarge",
"coreInstanceType" : "m3.xlarge",
"coreInstanceCount" : "10",
"taskInstanceType" : "m3.xlarge",
"taskInstanceCount": "10",
"bootstrapAction" : ["s3://Region.elasticmapreduce/bootstrap-actions/configure-hadoop,arg1,arg2,arg3","s3://Region.elasticmapreduce/bootstrap-actions/configure-hadoop/configure-other-stuff,arg1,arg2"]
}
```
# 启动具有版本标签 emr-4.x 或更高版本的 Amazon EMR 集群
**Example**
以下示例使用较新的 `releaseLabel` 字段启动 Amazon EMR 集群：
```
{
"id" : "MyEmrCluster",
"type" : "EmrCluster",
"keyPair" : "my-key-pair",
"masterInstanceType" : "m3.xlarge",
"coreInstanceType" : "m3.xlarge",
"coreInstanceCount" : "10",
"taskInstanceType" : "m3.xlarge",
"taskInstanceCount": "10",
"releaseLabel": "emr-4.1.0",
"applications": ["spark", "hive", "pig"],
"configuration": {"ref":"myConfiguration"}
}
```
# 在您的 Amazon EMR 集群上安装额外的软件
**Example**
`EmrCluster` 提供了 `supportedProducts` 字段，它在 Amazon EMR 集群上安装第三方软件，例如，它用于安装 Hadoop 自定义分发版本（如 MapR）。它接受适用于第三方软件读取和处理的参数的逗号分隔列表。以下示例说明如何使用 `EmrCluster` 的 `supportedProducts` 字段来创建已安装 Karmasphere Analytics 的自定义 MapR M3 版本集群，并在该集群上运行 `EmrActivity` 对象。
```
{
"id": "MyEmrActivity",
"type": "EmrActivity",
"schedule": {"ref": "ResourcePeriod"},
"runsOn": {"ref": "MyEmrCluster"},
"postStepCommand": "echo Ending job >> /mnt/var/log/stepCommand.txt",
"preStepCommand": "echo Starting job > /mnt/var/log/stepCommand.txt",
"step": "/home/hadoop/contrib/streaming/hadoop-streaming.jar,-input,s3n://elasticmapreduce/samples/wordcount/input,-output, \
hdfs:///output32113/,-mapper,s3n://elasticmapreduce/samples/wordcount/wordSplitter.py,-reducer,aggregate"
},
{
"id": "MyEmrCluster",
"type": "EmrCluster",
"schedule": {"ref": "ResourcePeriod"},
"supportedProducts": ["mapr,--edition,m3,--version,1.2,--key1,value1","karmasphere-enterprise-utility"],
"masterInstanceType": "m3.xlarge",
"taskInstanceType": "m3.xlarge"
}
```
# 在 3.x 版本上禁用服务器端加密
**Example**
默认情况下，由 Hadoop 2.x 版本创建的`EmrCluster`活动 AWS Data Pipeline 启用服务器端加密。如果您想禁用服务器端加密，则必须在集群对象定义中指定引导操作。
以下示例创建一个已禁用服务器端加密的 `EmrCluster` 活动：
```
{
"id":"NoSSEEmrCluster",
"type":"EmrCluster",
"hadoopVersion":"2.x",
"keyPair":"my-key-pair",
"masterInstanceType":"m3.xlarge",
"coreInstanceType":"m3.large",
"coreInstanceCount":"10",
"taskInstanceType":"m3.large",
"taskInstanceCount":"10",
"bootstrapAction":["s3://Region.elasticmapreduce/bootstrap-actions/configure-hadoop,-e, fs.s3.enableServerSideEncryption=false"]
}
```
# 在 4.x 版本上禁用服务器端加密
**Example**
您必须使用 `EmrConfiguration` 对象禁用服务器端加密。
以下示例创建一个已禁用服务器端加密的 `EmrCluster` 活动：
```
{
"name": "ReleaseLabelCluster",
"releaseLabel": "emr-4.1.0",
"applications": ["spark", "hive", "pig"],
"id": "myResourceId",
"type": "EmrCluster",
"configuration": {
"ref": "disableSSE"
}
},
{
"name": "disableSSE",
"id": "disableSSE",
"type": "EmrConfiguration",
"classification": "emrfs-site",
"property": [{
"ref": "enableServerSideEncryption"
}
]
},
{
"name": "enableServerSideEncryption",
"id": "enableServerSideEncryption",
"type": "Property",
"key": "fs.s3.enableServerSideEncryption",
"value": "false"
}
```
# 配置 Hadoop KMS ACLs 并在 HDFS 中创建加密区域
**Example**
以下对象是 ACLs 为 Hadoop KMS 创建的，并在 HDFS 中创建加密区域和相应的加密密钥：
```
{
"name": "kmsAcls",
"id": "kmsAcls",
"type": "EmrConfiguration",
"classification": "hadoop-kms-acls",
"property": [
{"ref":"kmsBlacklist"},
{"ref":"kmsAcl"}
]
},
{
"name": "hdfsEncryptionZone",
"id": "hdfsEncryptionZone",
"type": "EmrConfiguration",
"classification": "hdfs-encryption-zones",
"property": [
{"ref":"hdfsPath1"},
{"ref":"hdfsPath2"}
]
},
{
"name": "kmsBlacklist",
"id": "kmsBlacklist",
"type": "Property",
"key": "hadoop.kms.blacklist.CREATE",
"value": "foo,myBannedUser"
},
{
"name": "kmsAcl",
"id": "kmsAcl",
"type": "Property",
"key": "hadoop.kms.acl.ROLLOVER",
"value": "myAllowedUser"
},
{
"name": "hdfsPath1",
"id": "hdfsPath1",
"type": "Property",
"key": "/myHDFSPath1",
"value": "path1_key"
},
{
"name": "hdfsPath2",
"id": "hdfsPath2",
"type": "Property",
"key": "/myHDFSPath2",
"value": "path2_key"
}
```
# 指定自定义 IAM 角色
**Example**
默认情况下，`DataPipelineDefaultRole`作`DataPipelineDefaultResourceRole`为 Amazon EMR 服务角色和 Amazon EC2 实例配置文件 AWS Data Pipeline 传递以代表您创建资源。但是，您可以创建自定义 Amazon EMR 服务角色和自定义实例配置文件，然后改为使用它们。 AWS Data Pipeline 应有足够的权限使用自定义角色创建集群，并且必须添加 AWS Data Pipeline 为可信实体。
以下示例对象指定 Amazon EMR 集群的自定义角色：
```
{
"id":"MyEmrCluster",
"type":"EmrCluster",
"hadoopVersion":"2.x",
"keyPair":"my-key-pair",
"masterInstanceType":"m3.xlarge",
"coreInstanceType":"m3.large",
"coreInstanceCount":"10",
"taskInstanceType":"m3.large",
"taskInstanceCount":"10",
"role":"emrServiceRole",
"resourceRole":"emrInstanceProfile"
}
```
# 使用适用于 Java 的 AWS 开发工具包中的 EmrCluster 资源
**Example**
以下示例说明了如何使用 `EmrCluster` 和 `EmrActivity` 创建 Amazon EMR 4.x 集群以通过 Java 软件开发工具包运行 Spark 步骤：
```
public class dataPipelineEmr4 {
public static void main(String[] args) {
AWSCredentials credentials = null;
credentials = new ProfileCredentialsProvider("/path/to/AwsCredentials.properties","default").getCredentials();
DataPipelineClient dp = new DataPipelineClient(credentials);
CreatePipelineRequest createPipeline = new CreatePipelineRequest().withName("EMR4SDK").withUniqueId("unique");
CreatePipelineResult createPipelineResult = dp.createPipeline(createPipeline);
String pipelineId = createPipelineResult.getPipelineId();
PipelineObject emrCluster = new PipelineObject()
.withName("EmrClusterObj")
.withId("EmrClusterObj")
.withFields(
new Field().withKey("releaseLabel").withStringValue("emr-4.1.0"),
new Field().withKey("coreInstanceCount").withStringValue("3"),
new Field().withKey("applications").withStringValue("spark"),
new Field().withKey("applications").withStringValue("Presto-Sandbox"),
new Field().withKey("type").withStringValue("EmrCluster"),
new Field().withKey("keyPair").withStringValue("myKeyName"),
new Field().withKey("masterInstanceType").withStringValue("m3.xlarge"),
new Field().withKey("coreInstanceType").withStringValue("m3.xlarge")
);
PipelineObject emrActivity = new PipelineObject()
.withName("EmrActivityObj")
.withId("EmrActivityObj")
.withFields(
new Field().withKey("step").withStringValue("command-runner.jar,spark-submit,--executor-memory,1g,--class,org.apache.spark.examples.SparkPi,/usr/lib/spark/lib/spark-examples.jar,10"),
new Field().withKey("runsOn").withRefValue("EmrClusterObj"),
new Field().withKey("type").withStringValue("EmrActivity")
);
PipelineObject schedule = new PipelineObject()
.withName("Every 15 Minutes")
.withId("DefaultSchedule")
.withFields(
new Field().withKey("type").withStringValue("Schedule"),
new Field().withKey("period").withStringValue("15 Minutes"),
new Field().withKey("startAt").withStringValue("FIRST_ACTIVATION_DATE_TIME")
);
PipelineObject defaultObject = new PipelineObject()
.withName("Default")
.withId("Default")
.withFields(
new Field().withKey("failureAndRerunMode").withStringValue("CASCADE"),
new Field().withKey("schedule").withRefValue("DefaultSchedule"),
new Field().withKey("resourceRole").withStringValue("DataPipelineDefaultResourceRole"),
new Field().withKey("role").withStringValue("DataPipelineDefaultRole"),
new Field().withKey("pipelineLogUri").withStringValue("s3://myLogUri"),
new Field().withKey("scheduleType").withStringValue("cron")
);
List pipelineObjects = new ArrayList();
pipelineObjects.add(emrActivity);
pipelineObjects.add(emrCluster);
pipelineObjects.add(defaultObject);
pipelineObjects.add(schedule);
PutPipelineDefinitionRequest putPipelineDefintion = new PutPipelineDefinitionRequest()
.withPipelineId(pipelineId)
.withPipelineObjects(pipelineObjects);
PutPipelineDefinitionResult putPipelineResult = dp.putPipelineDefinition(putPipelineDefintion);
System.out.println(putPipelineResult);
ActivatePipelineRequest activatePipelineReq = new ActivatePipelineRequest()
.withPipelineId(pipelineId);
ActivatePipelineResult activatePipelineRes = dp.activatePipeline(activatePipelineReq);
System.out.println(activatePipelineRes);
System.out.println(pipelineId);
}
}
```
# 在私有子网中配置 Amazon EMR 集群
**Example**
该示例包括在 VPC 的私有子网中启动集群的配置。有关更多信息，请参阅 *Amazon EMR 管理指南*中的[在 VPC 中启动 Amazon EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-vpc-launching-job-flows.html)。此配置为可选配置。您可以在使用 `EmrCluster` 对象的任何管道中使用它。
要在私有子网中启动 Amazon EMR 集群，请在您的 `SubnetId` 配置中指定 `emrManagedMasterSecurityGroupId`、`emrManagedSlaveSecurityGroupId`、`serviceAccessSecurityGroupId`、和 `EmrCluster`。
```
{
"objects": [
{
"output": {
"ref": "S3BackupLocation"
},
"input": {
"ref": "DDBSourceTable"
},
"maximumRetries": "2",
"name": "TableBackupActivity",
"step": "s3://dynamodb-emr-#{myDDBRegion}/emr-ddb-storage-handler/2.1.0/emr-ddb-2.1.0.jar,org.apache.hadoop.dynamodb.tools.DynamoDbExport,#{output.directoryPath},#{input.tableName},#{input.readThroughputPercent}",
"id": "TableBackupActivity",
"runsOn": {
"ref": "EmrClusterForBackup"
},
"type": "EmrActivity",
"resizeClusterBeforeRunning": "false"
},
{
"readThroughputPercent": "#{myDDBReadThroughputRatio}",
"name": "DDBSourceTable",
"id": "DDBSourceTable",
"type": "DynamoDBDataNode",
"tableName": "#{myDDBTableName}"
},
{
"directoryPath": "#{myOutputS3Loc}/#{format(@scheduledStartTime, 'YYYY-MM-dd-HH-mm-ss')}",
"name": "S3BackupLocation",
"id": "S3BackupLocation",
"type": "S3DataNode"
},
{
"name": "EmrClusterForBackup",
"coreInstanceCount": "1",
"taskInstanceCount": "1",
"taskInstanceType": "m4.xlarge",
"coreInstanceType": "m4.xlarge",
"releaseLabel": "emr-4.7.0",
"masterInstanceType": "m4.xlarge",
"id": "EmrClusterForBackup",
"subnetId": "#{mySubnetId}",
"emrManagedMasterSecurityGroupId": "#{myMasterSecurityGroup}",
"emrManagedSlaveSecurityGroupId": "#{mySlaveSecurityGroup}",
"serviceAccessSecurityGroupId": "#{myServiceAccessSecurityGroup}",
"region": "#{myDDBRegion}",
"type": "EmrCluster",
"keyPair": "user-key-pair"
},
{
"failureAndRerunMode": "CASCADE",
"resourceRole": "DataPipelineDefaultResourceRole",
"role": "DataPipelineDefaultRole",
"pipelineLogUri": "#{myPipelineLogUri}",
"scheduleType": "ONDEMAND",
"name": "Default",
"id": "Default"
}
],
"parameters": [
{
"description": "Output S3 folder",
"id": "myOutputS3Loc",
"type": "AWS::S3::ObjectKey"
},
{
"description": "Source DynamoDB table name",
"id": "myDDBTableName",
"type": "String"
},
{
"default": "0.25",
"watermark": "Enter value between 0.1-1.0",
"description": "DynamoDB read throughput ratio",
"id": "myDDBReadThroughputRatio",
"type": "Double"
},
{
"default": "us-east-1",
"watermark": "us-east-1",
"description": "Region of the DynamoDB table",
"id": "myDDBRegion",
"type": "String"
}
],
"values": {
"myDDBRegion": "us-east-1",
"myDDBTableName": "ddb_table",
"myDDBReadThroughputRatio": "0.25",
"myOutputS3Loc": "s3://s3_path",
"mySubnetId": "subnet_id",
"myServiceAccessSecurityGroup": "service access security group",
"mySlaveSecurityGroup": "slave security group",
"myMasterSecurityGroup": "master security group",
"myPipelineLogUri": "s3://s3_path"
}
}
```
# 将 EBS 卷附加到集群节点
**Example**
您可以将 EBS 卷附加到您的管道内的 EMR 集群中的任何类型的节点。要将 EBS 卷附加到节点，请在您的 `EmrCluster` 配置中使用 `coreEbsConfiguration`、`masterEbsConfiguration` 和 `TaskEbsConfiguration`。
该 Amazon EMR 集群示例使用 Amazon EBS 卷作为其主节点、任务节点和核心节点。有关更多信息，请参阅 *Amazon EMR 管理指南*中的 [Amazon EMR 中的 Amazon EBS 卷](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-storage.html)。
这些配置是可选的。您可以在使用 `EmrCluster` 对象的任何管道中使用它们。
在管道中，单击 `EmrCluster` 对象配置，选择**主 EBS 配置**、**核心 EBS 配置**或**任务 EBS 配置**，然后输入类似于以下示例的配置详细信息。
```
{
"objects": [
{
"output": {
"ref": "S3BackupLocation"
},
"input": {
"ref": "DDBSourceTable"
},
"maximumRetries": "2",
"name": "TableBackupActivity",
"step": "s3://dynamodb-emr-#{myDDBRegion}/emr-ddb-storage-handler/2.1.0/emr-ddb-2.1.0.jar,org.apache.hadoop.dynamodb.tools.DynamoDbExport,#{output.directoryPath},#{input.tableName},#{input.readThroughputPercent}",
"id": "TableBackupActivity",
"runsOn": {
"ref": "EmrClusterForBackup"
},
"type": "EmrActivity",
"resizeClusterBeforeRunning": "false"
},
{
"readThroughputPercent": "#{myDDBReadThroughputRatio}",
"name": "DDBSourceTable",
"id": "DDBSourceTable",
"type": "DynamoDBDataNode",
"tableName": "#{myDDBTableName}"
},
{
"directoryPath": "#{myOutputS3Loc}/#{format(@scheduledStartTime, 'YYYY-MM-dd-HH-mm-ss')}",
"name": "S3BackupLocation",
"id": "S3BackupLocation",
"type": "S3DataNode"
},
{
"name": "EmrClusterForBackup",
"coreInstanceCount": "1",
"taskInstanceCount": "1",
"taskInstanceType": "m4.xlarge",
"coreInstanceType": "m4.xlarge",
"releaseLabel": "emr-4.7.0",
"masterInstanceType": "m4.xlarge",
"id": "EmrClusterForBackup",
"subnetId": "#{mySubnetId}",
"emrManagedMasterSecurityGroupId": "#{myMasterSecurityGroup}",
"emrManagedSlaveSecurityGroupId": "#{mySlaveSecurityGroup}",
"region": "#{myDDBRegion}",
"type": "EmrCluster",
"coreEbsConfiguration": {
"ref": "EBSConfiguration"
},
"masterEbsConfiguration": {
"ref": "EBSConfiguration"
},
"taskEbsConfiguration": {
"ref": "EBSConfiguration"
},
"keyPair": "user-key-pair"
},
{
"name": "EBSConfiguration",
"id": "EBSConfiguration",
"ebsOptimized": "true",
"ebsBlockDeviceConfig" : [
{ "ref": "EbsBlockDeviceConfig" }
],
"type": "EbsConfiguration"
},
{
"name": "EbsBlockDeviceConfig",
"id": "EbsBlockDeviceConfig",
"type": "EbsBlockDeviceConfig",
"volumesPerInstance" : "2",
"volumeSpecification" : {
"ref": "VolumeSpecification"
}
},
{
"name": "VolumeSpecification",
"id": "VolumeSpecification",
"type": "VolumeSpecification",
"sizeInGB": "500",
"volumeType": "io1",
"iops": "1000"
},
{
"failureAndRerunMode": "CASCADE",
"resourceRole": "DataPipelineDefaultResourceRole",
"role": "DataPipelineDefaultRole",
"pipelineLogUri": "#{myPipelineLogUri}",
"scheduleType": "ONDEMAND",
"name": "Default",
"id": "Default"
}
],
"parameters": [
{
"description": "Output S3 folder",
"id": "myOutputS3Loc",
"type": "AWS::S3::ObjectKey"
},
{
"description": "Source DynamoDB table name",
"id": "myDDBTableName",
"type": "String"
},
{
"default": "0.25",
"watermark": "Enter value between 0.1-1.0",
"description": "DynamoDB read throughput ratio",
"id": "myDDBReadThroughputRatio",
"type": "Double"
},
{
"default": "us-east-1",
"watermark": "us-east-1",
"description": "Region of the DynamoDB table",
"id": "myDDBRegion",
"type": "String"
}
],
"values": {
"myDDBRegion": "us-east-1",
"myDDBTableName": "ddb_table",
"myDDBReadThroughputRatio": "0.25",
"myOutputS3Loc": "s3://s3_path",
"mySubnetId": "subnet_id",
"mySlaveSecurityGroup": "slave security group",
"myMasterSecurityGroup": "master security group",
"myPipelineLogUri": "s3://s3_path"
}
}
```