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 ⇒ JobFlow

Returns a new instance of JobFlow.



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/elasticity/job_flow.rb', line 29

def initialize
  @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

  @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'
end

Instance Attribute Details

#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.



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

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.



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

def ec2_subnet_id
  @ec2_subnet_id
end

#enable_debugging ⇒ Object

Returns the value of attribute enable_debugging.



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

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.



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

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.



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

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.



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

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.



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

def placement
  @placement
end

#region ⇒ Object

Returns the value of attribute region.



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

def region
  @region
end

#service_role ⇒ Object

Returns the value of attribute service_role.



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

def service_role
  @service_role
end

#slave_instance_type ⇒ Object

Returns the value of attribute slave_instance_type.



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

def slave_instance_type
  @slave_instance_type
end

#tags ⇒ Object

Returns the value of attribute tags.



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

def tags
  @tags
end

#visible_to_all_users ⇒ Object

Returns the value of attribute visible_to_all_users.



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

def visible_to_all_users
  @visible_to_all_users
end

Class Method Details

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



51
52
53
54
55
56
57
 
# File 'lib/elasticity/job_flow.rb', line 51

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

Instance Method Details

#add_bootstrap_action(bootstrap_action) ⇒ Object 



101
102
103
104
105
106
 
# File 'lib/elasticity/job_flow.rb', line 101

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 



123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/elasticity/job_flow.rb', line 123

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, jobflow_steps)
  else
    @jobflow_steps << jobflow_step
  end
end

#cluster_status ⇒ Object 



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

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

#cluster_step_status ⇒ Object 



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

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

#run ⇒ Object 



136
137
138
139
140
141
 
# File 'lib/elasticity/job_flow.rb', line 136

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 



113
114
115
116
 
# File 'lib/elasticity/job_flow.rb', line 113

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

#set_master_instance_group(instance_group) ⇒ Object 



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

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

#set_task_instance_group(instance_group) ⇒ Object 



118
119
120
121
 
# File 'lib/elasticity/job_flow.rb', line 118

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

#shutdown ⇒ Object 



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

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

#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