Блог

Как я могу добавить шаги к ожидающему потоку заданий Amazon EMR с помощью boto без завершения потока заданий после завершения?

Я создал интерактивный поток заданий на Amazon Elastic Map Reduce и загрузил несколько таблиц. Когда я передаю новые шаги в поток заданий с помощью Boto emr_conn.add_jobflow_steps(...) , поток заданий завершается после его завершения или сбоя.

Я знаю, что могу запустить поток заданий с помощью boto run_jobflow с keep_alive помощью параметра — но я хотел бы работать с потоками, которые уже запущены.

Если он завершается правильно, он не должен завершаться keep_alive=True . Тем не менее, обычно он завершается при сбое, поэтому вы хотите добавить terminate_on_failure="CONTINUE" к своим add_job_steps аргументам.

Я использую что-то вроде этого

создайте с помощью

 import boto.emr

conn = boto.emr.connect_to_region('us-west-2')
jobid = conn.run_jobflow(name='cluster-name',
                     ec2_keyname="yourkeyhere",
                     num_instances=3,
                     master_instance_type='m1.medium',
                     slave_instance_type='m1.medium',
                     keep_alive=True,
)
 

добавьте задание в существующий кластер с помощью (подождите немного, пока кластер перейдет в состояние ожидания)

 import boto.emr

conn = boto.emr.connect_to_region('us-west-2')
# get the list of waiting cluster and take the first one
jobid =  conn.describe_jobflows(states=["WAITING"])[0].jobflowid
print jobid
flow_steps = list()
runThing = boto.emr.step.ScriptRunnerStep(
                        name="job step name",
                        step_args = ["s3://yours3bucket/dosmthg.sh"])
flow_steps.append(runThing)
conn.add_jobflow_steps(jobid, flow_steps)
 

Примечания

• вам необходимо заполнить ~/.aws/учетные данные (настройка aws)
• регион Amazon us-west-2 в настоящее время имеет более свежую версию ami
• возможно, вам придется добавить bootstrap_actions=, если вам нужны hive, pig или пользовательские шаги установки

Вы также можете сделать это с помощью флага ‘KeepJobFlowAliveWhenNoSteps’.

 response = emr.run_job_flow(
        Name="start-my-cluster",
        ReleaseLabel="emr-5.3.1",
        LogUri='s3://logs',
        Instances={
            'InstanceGroups': [
                {'Name': 'EmrMaster',
                 'InstanceRole': 'MASTER',
                 'InstanceType': 'm3.xlarge',
                 'InstanceCount': 1},
                {'Name': 'EmrCore',
                 'InstanceRole': 'CORE',
                 'InstanceType': 'm3.xlarge',
                 'InstanceCount': 2}
            ],
            'Ec2KeyName': 'my-key-name',
            'KeepJobFlowAliveWhenNoSteps' : True,
        },
        Applications=[{'Name': 'Hadoop'}, {'Name': 'Spark'}, {'Name': 'Hive'}],
        JobFlowRole='EMR_EC2_DefaultRole',
        ServiceRole='EMR_DefaultRole',
        VisibleToAllUsers=True,
        Steps=[
            # steps go here...
        ]
        )