1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
|
# Copyright 2011-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
require 'date'
require 'time'
module AWS
class EMR
# # Creating a Job Flow
#
# Call {#create} to run a new job flow.
#
# emr = AWS::EMR.new
#
# job_flow = emr.job_flows.create('name',
# :instances => {
# :instance_count => 2,
# :master_instance_type => 'm1.small',
# :slave_instance_type => 'm1.small',
# }
# )
#
# # Getting a Job Flow
#
# You can get a job flow by its ID.
#
# job_flow = emr.job_flows['j-123456678'] # makes no request
# job_flow.exists? #=> true/false
#
# # Enumerating Job Flows
#
# You can enumerate all job flows, or filter them.
#
# # all job flows
# job_flows.each {|job_flow| ... }
#
# # only job flows with a particular state
# job_flows.with_state('ENDED').each {|job_flow| ... }
#
# The filtering methods include:
#
# * {#with_id}
# * {#with_state}
# * {#created_before}
# * {#created_after}
#
class JobFlowCollection
include Core::Collection::Simple
# @api private
def initialize options = {}
@filters = options[:filters] || {}
super
end
# @param [String] job_flow_id
# @return [JobFlow] Returns a {JobFlow} with the given ID.
def [] job_flow_id
JobFlow.new(job_flow_id, :config => config)
end
# Runs a job flow.
#
# job_flow = emr.job_flows.create('name',
# :instances => {
# :instance_count => 2,
# :master_instance_type => 'm1.small',
# :slave_instance_type => 'm1.small',
# }
# )
#
# See {Client#run_job_flow} for documentation on the complete
# list of accepted options.
# @param [String] name
# @param [Hash] options
# @see (Client#run_job_flow)
# @return [JobFlow]
def create name, options = {}
options[:name] = name
options[:ami_version] ||= 'latest'
options[:instances] ||= {}
resp = client.run_job_flow(options)
self[resp.data[:job_flow_id]]
end
alias_method :run, :create
# Returns a new collection that will only enumerate job flows that have
# one of the given ids.
#
# emr.job_flows.with_id('id1', 'id2', 'id3').each do |job_flow|
# # ...
# end
#
# @param [String] ids One or more job flow ids to use as a filter.
# @return [JobFlowCollection]
def with_id *ids
filter(:job_flow_ids, ids.flatten)
end
# Returns a new collection that will only enumerate job flows that have
# one of the given job flow states.
#
# emr.job_flows.with_state('SHUTTING_DOWN', 'TERMINATED').each do |job|
# # ...
# end
#
# @param [String] states One or more job flow states to use as a filter.
# @return [JobFlowCollection]
def with_state *states
filter(:job_flow_states, states.flatten)
end
# Returns a new collection that will only enumerate job flows that
# were created before the given time.
#
# # enumerate jobs that are more than an hour old
# emr.job_flows.created_before(Time.now - 3600).each{|job| ... }
#
# @param [Time,DateTime,Date,Integer] time
# @return [JobFlowCollection]
def created_before time
filter(:created_before, time.to_i)
end
# Returns a new collection that will only enumerate job flows that
# were created after the given time.
#
# # enumerate jobs that are at most 1 hour old
# emr.job_flows.created_after(Time.now - 3600).each{|job| ... }
#
# @param [Time,DateTime,Date,Integer] time
# @return [JobFlowCollection]
def created_after time
filter(:created_after, time.to_i)
end
# @param [String,Symbol] name
# @param [Mixed] value
# @return [JobFlowCollection]
def filter name, value
options = {}
options[:filters] = @filters.merge(name.to_s.to_sym => value)
options[:config] = config
JobFlowCollection.new(options)
end
protected
def _each_item options = {}, &block
resp = client.describe_job_flows(@filters.merge(options))
resp.data[:job_flows].each do |details|
job_flow = JobFlow.new_from(
:describe_job_flows,
details,
details[:job_flow_id],
:config => config)
yield(job_flow)
end
end
end
end
end
|
