Class: Elasticity::JobFlow

Inherits:
Object
  • Object
show all
Defined in:
lib/elasticity/job_flow.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(access = nil, secret = nil) ⇒ JobFlow

Returns a new instance of JobFlow.



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/elasticity/job_flow.rb', line 31

def initialize(access=nil, secret=nil)
  @action_on_failure = 'TERMINATE_JOB_FLOW'
  @name = 'Elasticity Job Flow'
  @ami_version = 'latest'
  @keep_job_flow_alive_when_no_steps = false
  self.placement = 'us-east-1a'
  @enable_debugging = false

  @access_key = access
  @secret_key = secret
  @visible_to_all_users = false

  @bootstrap_actions = []
  @jobflow_steps = []
  @installed_steps = []

  @instance_groups = {}
  set_master_instance_group(Elasticity::InstanceGroup.new)
  set_core_instance_group(Elasticity::InstanceGroup.new)
  @instance_count = 2
  @master_instance_type = 'm1.small'
  @slave_instance_type = 'm1.small'

  @access_key = access
  @secret_key = secret
end

Instance Attribute Details

#access_key ⇒ Object (readonly)

Returns the value of attribute access_key.



28
29
30
 
# File 'lib/elasticity/job_flow.rb', line 28

def access_key
  @access_key
end

#action_on_failure ⇒ Object

Returns the value of attribute action_on_failure.



11
12
13
 
# File 'lib/elasticity/job_flow.rb', line 11

def action_on_failure
  @action_on_failure
end

#ami_version ⇒ Object

Returns the value of attribute ami_version.



18
19
20
 
# File 'lib/elasticity/job_flow.rb', line 18

def ami_version
  @ami_version
end

#ec2_key_name ⇒ Object

Returns the value of attribute ec2_key_name.



12
13
14
 
# File 'lib/elasticity/job_flow.rb', line 12

def ec2_key_name
  @ec2_key_name
end

#ec2_subnet_id ⇒ Object

Returns the value of attribute ec2_subnet_id.



20
21
22
 
# File 'lib/elasticity/job_flow.rb', line 20

def ec2_subnet_id
  @ec2_subnet_id
end

#enable_debugging ⇒ Object

Returns the value of attribute enable_debugging.



24
25
26
 
# File 'lib/elasticity/job_flow.rb', line 24

def enable_debugging
  @enable_debugging
end

#instance_count ⇒ Object

Returns the value of attribute instance_count.



14
15
16
 
# File 'lib/elasticity/job_flow.rb', line 14

def instance_count
  @instance_count
end

#job_flow_role ⇒ Object

Returns the value of attribute job_flow_role.



25
26
27
 
# File 'lib/elasticity/job_flow.rb', line 25

def job_flow_role
  @job_flow_role
end

#keep_job_flow_alive_when_no_steps ⇒ Object

Returns the value of attribute keep_job_flow_alive_when_no_steps.



19
20
21
 
# File 'lib/elasticity/job_flow.rb', line 19

def keep_job_flow_alive_when_no_steps
  @keep_job_flow_alive_when_no_steps
end

#log_uri ⇒ Object

Returns the value of attribute log_uri.



15
16
17
 
# File 'lib/elasticity/job_flow.rb', line 15

def log_uri
  @log_uri
end

#master_instance_type ⇒ Object

Returns the value of attribute master_instance_type.



16
17
18
 
# File 'lib/elasticity/job_flow.rb', line 16

def master_instance_type
  @master_instance_type
end

#name ⇒ Object

Returns the value of attribute name.



13
14
15
 
# File 'lib/elasticity/job_flow.rb', line 13

def name
  @name
end

#placement ⇒ Object

Returns the value of attribute placement.



21
22
23
 
# File 'lib/elasticity/job_flow.rb', line 21

def placement
  @placement
end

#region ⇒ Object

Returns the value of attribute region.



22
23
24
 
# File 'lib/elasticity/job_flow.rb', line 22

def region
  @region
end

#secret_key ⇒ Object (readonly)

Returns the value of attribute secret_key.



29
30
31
 
# File 'lib/elasticity/job_flow.rb', line 29

def secret_key
  @secret_key
end

#service_role ⇒ Object

Returns the value of attribute service_role.



26
27
28
 
# File 'lib/elasticity/job_flow.rb', line 26

def service_role
  @service_role
end

#slave_instance_type ⇒ Object

Returns the value of attribute slave_instance_type.



17
18
19
 
# File 'lib/elasticity/job_flow.rb', line 17

def slave_instance_type
  @slave_instance_type
end

#visible_to_all_users ⇒ Object

Returns the value of attribute visible_to_all_users.



23
24
25
 
# File 'lib/elasticity/job_flow.rb', line 23

def visible_to_all_users
  @visible_to_all_users
end

Class Method Details

.from_jobflow_id(access, secret, jobflow_id, region = 'us-east-1') ⇒ Object 



58
59
60
61
62
63
64
 
# File 'lib/elasticity/job_flow.rb', line 58

def self.from_jobflow_id(access, secret, jobflow_id, region = 'us-east-1')
  JobFlow.new(access, secret).tap do |j|
    j.instance_variable_set(:@region, region)
    j.instance_variable_set(:@jobflow_id, jobflow_id)
    j.instance_variable_set(:@installed_steps, j.status.installed_steps)
  end
end

Instance Method Details

#add_bootstrap_action(bootstrap_action) ⇒ Object 



108
109
110
111
112
113
 
# File 'lib/elasticity/job_flow.rb', line 108

def add_bootstrap_action(bootstrap_action)
  if is_jobflow_running?
    raise JobFlowRunningError, 'To modify bootstrap actions, please create a new job flow.'
  end
  @bootstrap_actions << bootstrap_action
end

#add_step(jobflow_step) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/elasticity/job_flow.rb', line 130

def add_step(jobflow_step)
  if is_jobflow_running?
    jobflow_steps = []
    if jobflow_step.requires_installation? && !@installed_steps.include?(jobflow_step.class)
      jobflow_steps.concat(jobflow_step.aws_installation_steps)
    end
    jobflow_steps << jobflow_step.to_aws_step(self)
    emr.add_jobflow_steps(@jobflow_id, {:steps => jobflow_steps})
  else
    @jobflow_steps << jobflow_step
  end
end

#run ⇒ Object 



143
144
145
146
147
148
 
# File 'lib/elasticity/job_flow.rb', line 143

def run
  if is_jobflow_running?
    raise JobFlowRunningError, 'Cannot run a job flow multiple times.  To do more with this job flow, please use #add_step.'
  end
  @jobflow_id = emr.run_job_flow(jobflow_config)
end

#set_core_instance_group(instance_group) ⇒ Object 



120
121
122
123
 
# File 'lib/elasticity/job_flow.rb', line 120

def set_core_instance_group(instance_group)
  instance_group.role = 'CORE'
  @instance_groups[:core] = instance_group
end

#set_master_instance_group(instance_group) ⇒ Object 



115
116
117
118
 
# File 'lib/elasticity/job_flow.rb', line 115

def set_master_instance_group(instance_group)
  instance_group.role = 'MASTER'
  @instance_groups[:master] = instance_group
end

#set_task_instance_group(instance_group) ⇒ Object 



125
126
127
128
 
# File 'lib/elasticity/job_flow.rb', line 125

def set_task_instance_group(instance_group)
  instance_group.role = 'TASK'
  @instance_groups[:task] = instance_group
end

#shutdown ⇒ Object 



150
151
152
153
154
155
 
# File 'lib/elasticity/job_flow.rb', line 150

def shutdown
  if !is_jobflow_running?
    raise JobFlowNotStartedError, 'Cannot #shutdown a job flow that has not yet been #run.'
  end
  emr.terminate_jobflows(@jobflow_id)
end

#status ⇒ Object 



157
158
159
160
161
162
 
# File 'lib/elasticity/job_flow.rb', line 157

def status
  if !is_jobflow_running?
    raise JobFlowNotStartedError, 'Please #run this job flow before attempting to retrieve status.'
  end
  emr.describe_jobflow(@jobflow_id)
end

#wait_for_completion(&on_wait) ⇒ Object 



164
165
166
167
 
# File 'lib/elasticity/job_flow.rb', line 164

def wait_for_completion(&on_wait)
  l = Elasticity::Looper.new(method(:retry_check), on_wait)
  l.go
end