我正在尝试启动集群并使用boto运行所有作业。我找到了许多创建job_flows的例子。但我不能为我的生活找到一个例子,表明:
我错过了什么吗?
Boto和底层的EMR API目前正在混合术语集群和作业流程,而工作流程正在deprecated。我认为他们是同义词。
您可以通过调用boto.emr.connection.run_jobflow()
函数来创建新集群。它将返回EMR为您生成的集群ID。
首先是所有强制性的东西:
#!/usr/bin/env python
import boto
import boto.emr
from boto.emr.instance_group import InstanceGroup
conn = boto.emr.connect_to_region('us-east-1')
然后我们指定实例组,包括我们要为TASK节点支付的现货价格:
instance_groups = []
instance_groups.append(InstanceGroup(
num_instances=1,
role="MASTER",
type="m1.small",
market="ON_DEMAND",
name="Main node"))
instance_groups.append(InstanceGroup(
num_instances=2,
role="CORE",
type="m1.small",
market="ON_DEMAND",
name="Worker nodes"))
instance_groups.append(InstanceGroup(
num_instances=2,
role="TASK",
type="m1.small",
market="SPOT",
name="My cheap spot nodes",
bidprice="0.002"))
最后我们开始一个新的集群:
cluster_id = conn.run_jobflow(
"Name for my cluster",
instance_groups=instance_groups,
action_on_failure='TERMINATE_JOB_FLOW',
keep_alive=True,
enable_debugging=True,
log_uri="s3://mybucket/logs/",
hadoop_version=None,
ami_version="2.4.9",
steps=[],
bootstrap_actions=[],
ec2_keyname="my-ec2-key",
visible_to_all_users=True,
job_flow_role="EMR_EC2_DefaultRole",
service_role="EMR_DefaultRole")
如果我们关心,我们也可以打印集群ID:
print "Starting cluster", cluster_id
我相信使用boto3启动EMR集群的最小Python数量是:
import boto3
client = boto3.client('emr', region_name='us-east-1')
response = client.run_job_flow(
Name="Boto3 test cluster",
ReleaseLabel='emr-5.12.0',
Instances={
'MasterInstanceType': 'm4.xlarge',
'SlaveInstanceType': 'm4.xlarge',
'InstanceCount': 3,
'KeepJobFlowAliveWhenNoSteps': True,
'TerminationProtected': False,
'Ec2SubnetId': 'my-subnet-id',
'Ec2KeyName': 'my-key',
},
VisibleToAllUsers=True,
JobFlowRole='EMR_EC2_DefaultRole',
ServiceRole='EMR_DefaultRole'
)
注意:你必须要create EMR_EC2_DefaultRole
and EMR_DefaultRole
。 Amazon documentation声称JobFlowRole
和ServiceRole
是可选的,但省略它们对我不起作用。这可能是因为我的子网是VPC子网,但我不确定。
我使用以下代码创建安装了flink的EMR,并包含3个实例组。参考文件:https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/emr.html#EMR.Client.run_job_flow
import boto3
masterInstanceType = 'm4.large'
coreInstanceType = 'c3.xlarge'
taskInstanceType = 'm4.large'
coreInstanceNum = 2
taskInstanceNum = 2
clusterName = 'my-emr-name'
emrClient = boto3.client('emr')
logUri = 's3://bucket/xxxxxx/'
releaseLabel = 'emr-5.17.0' #emr version
instances = {
'Ec2KeyName': 'my_keyxxxxxx',
'Ec2SubnetId': 'subnet-xxxxxx',
'ServiceAccessSecurityGroup': 'sg-xxxxxx',
'EmrManagedMasterSecurityGroup': 'sg-xxxxxx',
'EmrManagedSlaveSecurityGroup': 'sg-xxxxxx',
'KeepJobFlowAliveWhenNoSteps': True,
'TerminationProtected': False,
'InstanceGroups': [{
'InstanceRole': 'MASTER',
"InstanceCount": 1,
"InstanceType": masterInstanceType,
"Market": "SPOT",
"Name": "Master"
}, {
'InstanceRole': 'CORE',
"InstanceCount": coreInstanceNum,
"InstanceType": coreInstanceType,
"Market": "SPOT",
"Name": "Core",
}, {
'InstanceRole': 'TASK',
"InstanceCount": taskInstanceNum,
"InstanceType": taskInstanceType,
"Market": "SPOT",
"Name": "Core",
}
]
}
bootstrapActions = [{
'Name': 'Log to Cloudwatch Logs',
'ScriptBootstrapAction': {
'Path': 's3://mybucket/bootstrap_cwl.sh'
}
}, {
'Name': 'Custom action',
'ScriptBootstrapAction': {
'Path': 's3://mybucket/install.sh'
}
}]
applications = [{'Name': 'Flink'}]
serviceRole = 'EMR_DefaultRole'
jobFlowRole = 'EMR_EC2_DefaultRole'
tags = [{'Key': 'keyxxxxxx', 'Value': 'valuexxxxxx'},
{'Key': 'key2xxxxxx', 'Value': 'value2xxxxxx'}
]
steps = [
{
'Name': 'Run Flink',
'ActionOnFailure': 'TERMINATE_JOB_FLOW',
'HadoopJarStep': {
'Jar': 'command-runner.jar',
'Args': ['flink', 'run',
'-m', 'yarn-cluster',
'-p', str(taskInstanceNum),
'-yjm', '1024',
'-ytm', '1024',
'/home/hadoop/test-1.0-SNAPSHOT.jar'
]
}
},
]
response = emrClient.run_job_flow(
Name=clusterName,
LogUri=logUri,
ReleaseLabel=releaseLabel,
Instances=instances,
Steps=steps,
Configurations=configurations,
BootstrapActions=bootstrapActions,
Applications=applications,
ServiceRole=serviceRole,
JobFlowRole=jobFlowRole,
Tags=tags
)
我的步骤论点是:bash -c /usr/bin/flink run -m yarn-cluster -yn 2 /home/hadoop/mysflinkjob.jar
尝试执行相同的run_job_flow,但收到错误:
无法运行程序“/ usr / bin / flink run -m yarn-cluster -yn 2 /home/hadoop/mysflinkjob.jar”(在目录“。”中):error = 2,没有这样的文件或目录
从主节点执行相同的命令工作正常,但不是从Python boto3
似乎问题是由EMR或boto3添加到参数中的引号引起的。
更新:
用白色空格拆分所有参数。我的意思是,如果你需要执行"flink run myflinkjob.jar"
传递你的参数作为这个列表:
[轻快 “” 运行 “” myflinkjob.jar]