YARN原理分析

  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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382

public interface YarnScheduler extends EventHandler<SchedulerEvent> {

  /**
   * Get queue information.
   *
   * @param queueName queue name
   * @param includeChildQueues include child queues?
   * @param recursive get children queues?
   * @return queue information
   * @throws IOException an I/O exception has occurred.
   */
  @Public
  @Stable
  public QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
      boolean recursive) throws IOException;

  /**
   * Get acls for queues for current user.
   * @return acls for queues for current user
   */
  @Public
  @Stable
  public List<QueueUserACLInfo> getQueueUserAclInfo();

  /**
   * Get the whole resource capacity of the cluster.
   * @return the whole resource capacity of the cluster.
   */
  @LimitedPrivate("yarn")
  @Unstable
  public Resource getClusterResource();

  /**
   * Get minimum allocatable {@link Resource}.
   * @return minimum allocatable resource
   */
  @Public
  @Stable
  public Resource getMinimumResourceCapability();
  
  /**
   * Get maximum allocatable {@link Resource} at the cluster level.
   * @return maximum allocatable resource
   */
  @Public
  @Stable
  public Resource getMaximumResourceCapability();

  /**
   * Get maximum allocatable {@link Resource} for the queue specified.
   * @param queueName queue name
   * @return maximum allocatable resource
   */
  @Public
  @Stable
  public Resource getMaximumResourceCapability(String queueName);

  @LimitedPrivate("yarn")
  @Evolving
  ResourceCalculator getResourceCalculator();

  /**
   * Get the number of nodes available in the cluster.
   * @return the number of available nodes.
   */
  @Public
  @Stable
  public int getNumClusterNodes();
  
  /**
   * The main API between the ApplicationMaster and the Scheduler.
   * The ApplicationMaster may request/update container resources,
   * number of containers, node/rack preference for allocations etc.
   * to the Scheduler.
   * @param appAttemptId the id of the application attempt.
   * @param ask the request made by an application to obtain various allocations
   * like host/rack, resource, number of containers, relaxLocality etc.,
   * see {@link ResourceRequest}.
   * @param schedulingRequests similar to ask, but with added ability to specify
   * allocation tags etc., see {@link SchedulingRequest}.
   * @param release the list of containers to be released.
   * @param blacklistAdditions places (node/rack) to be added to the blacklist.
   * @param blacklistRemovals places (node/rack) to be removed from the
   * blacklist.
   * @param updateRequests container promotion/demotion updates.
   * @return the {@link Allocation} for the application.
   */
  @Public
  @Stable
  Allocation allocate(ApplicationAttemptId appAttemptId,
      List<ResourceRequest> ask, List<SchedulingRequest> schedulingRequests,
      List<ContainerId> release, List<String> blacklistAdditions,
      List<String> blacklistRemovals, ContainerUpdates updateRequests);

  /**
   * Get node resource usage report.
   *
   * @param nodeId nodeId.
   * @return the {@link SchedulerNodeReport} for the node or null
   * if nodeId does not point to a defined node.
   */
  @LimitedPrivate("yarn")
  @Stable
  public SchedulerNodeReport getNodeReport(NodeId nodeId);
  
  /**
   * Get the Scheduler app for a given app attempt Id.
   * @param appAttemptId the id of the application attempt
   * @return SchedulerApp for this given attempt.
   */
  @LimitedPrivate("yarn")
  @Stable
  SchedulerAppReport getSchedulerAppInfo(ApplicationAttemptId appAttemptId);

  /**
   * Get a resource usage report from a given app attempt ID.
   * @param appAttemptId the id of the application attempt
   * @return resource usage report for this given attempt
   */
  @LimitedPrivate("yarn")
  @Evolving
  ApplicationResourceUsageReport getAppResourceUsageReport(
      ApplicationAttemptId appAttemptId);
  
  /**
   * Get the root queue for the scheduler.
   * @return the root queue for the scheduler.
   */
  @LimitedPrivate("yarn")
  @Evolving
  QueueMetrics getRootQueueMetrics();

  /**
   * Check if the user has permission to perform the operation.
   * If the user has {@link QueueACL#ADMINISTER_QUEUE} permission,
   * this user can view/modify the applications in this queue.
   *
   * @param callerUGI caller UserGroupInformation.
   * @param acl queue ACL.
   * @param queueName queue Name.
   * @return <code>true</code> if the user has the permission,
   *         <code>false</code> otherwise
   */
  boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName);
  
  /**
   * Gets the apps under a given queue
   * @param queueName the name of the queue.
   * @return a collection of app attempt ids in the given queue.
   */
  @LimitedPrivate("yarn")
  @Stable
  public List<ApplicationAttemptId> getAppsInQueue(String queueName);

  /**
   * Get the container for the given containerId.
   *
   * @param containerId the given containerId.
   * @return the container for the given containerId.
   */
  @LimitedPrivate("yarn")
  @Unstable
  public RMContainer getRMContainer(ContainerId containerId);

  /**
   * Moves the given application to the given queue.
   * @param appId application Id
   * @param newQueue the given queue.
   * @return the name of the queue the application was placed into
   * @throws YarnException if the move cannot be carried out
   */
  @LimitedPrivate("yarn")
  @Evolving
  public String moveApplication(ApplicationId appId, String newQueue)
      throws YarnException;

  /**
   *
   * @param appId Application ID
   * @param newQueue Target QueueName
   * @throws YarnException if the pre-validation for move cannot be carried out
   */
  @LimitedPrivate("yarn")
  @Evolving
  public void preValidateMoveApplication(ApplicationId appId,
      String newQueue) throws YarnException;

  /**
   * Completely drain sourceQueue of applications, by moving all of them to
   * destQueue.
   *
   * @param sourceQueue sourceQueue.
   * @param destQueue destQueue.
   * @throws YarnException when yarn exception occur.
   */
  void moveAllApps(String sourceQueue, String destQueue) throws YarnException;

  /**
   * Terminate all applications in the specified queue.
   *
   * @param queueName the name of queue to be drained
   * @throws YarnException when yarn exception occur.
   */
  void killAllAppsInQueue(String queueName) throws YarnException;

  /**
   * Remove an existing queue. Implementations might limit when a queue could be
   * removed (e.g., must have zero entitlement, and no applications running, or
   * must be a leaf, etc..).
   *
   * @param queueName name of the queue to remove
   * @throws YarnException when yarn exception occur.
   */
  void removeQueue(String queueName) throws YarnException;

  /**
   * Add to the scheduler a new Queue. Implementations might limit what type of
   * queues can be dynamically added (e.g., Queue must be a leaf, must be
   * attached to existing parent, must have zero entitlement).
   *
   * @param newQueue the queue being added.
   * @throws YarnException when yarn exception occur.
   * @throws IOException when io exception occur.
   */
  void addQueue(Queue newQueue) throws YarnException, IOException;

  /**
   * This method increase the entitlement for current queue (must respect
   * invariants, e.g., no overcommit of parents, non negative, etc.).
   * Entitlement is a general term for weights in FairScheduler, capacity for
   * the CapacityScheduler, etc.
   *
   * @param queue the queue for which we change entitlement
   * @param entitlement the new entitlement for the queue (capacity,
   *              maxCapacity, etc..)
   * @throws YarnException when yarn exception occur.
   */
  void setEntitlement(String queue, QueueEntitlement entitlement)
      throws YarnException;

  /**
   * Gets the list of names for queues managed by the Reservation System.
   * @return the list of queues which support reservations
   * @throws YarnException when yarn exception occur.
   */
  public Set<String> getPlanQueues() throws YarnException;  

  /**
   * Return a collection of the resource types that are considered when
   * scheduling
   *
   * @return an EnumSet containing the resource types
   */
  public EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes();

  /**
   *
   * Verify whether a submitted application priority is valid as per configured
   * Queue
   *
   * @param priorityRequestedByApp
   *          Submitted Application priority.
   * @param user
   *          User who submitted the Application
   * @param queuePath
   *          Name of the Queue
   * @param applicationId
   *          Application ID
   * @return Updated Priority from scheduler
   * @throws YarnException when yarn exception occur.
   */
  public Priority checkAndGetApplicationPriority(Priority priorityRequestedByApp,
      UserGroupInformation user, String queuePath, ApplicationId applicationId)
      throws YarnException;

  /**
   *
   * Change application priority of a submitted application at runtime
   *
   * @param newPriority Submitted Application priority.
   *
   * @param applicationId Application ID
   *
   * @param future Sets any type of exception happened from StateStore
   * @param user who submitted the application
   *
   * @return updated priority
   * @throws YarnException when yarn exception occur.
   */
  public Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user) throws YarnException;

  /**
   *
   * Get previous attempts' live containers for work-preserving AM restart.
   *
   * @param appAttemptId the id of the application attempt
   *
   * @return list of live containers for the given attempt
   */
  List<Container> getTransferredContainers(ApplicationAttemptId appAttemptId);

  /**
   * Set the cluster max priority.
   * 
   * @param conf Configuration.
   * @throws YarnException when yarn exception occur.
   */
  void setClusterMaxPriority(Configuration conf) throws YarnException;

  /**
   * Get pending resource request for specified application attempt.
   *
   * @param attemptId the id of the application attempt
   * @return pending resource requests.
   */
  List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId);

  /**
   * Get pending scheduling request for specified application attempt.
   *
   * @param attemptId the id of the application attempt
   *
   * @return pending scheduling requests
   */
  List<SchedulingRequest> getPendingSchedulingRequestsForAttempt(
      ApplicationAttemptId attemptId);

  /**
   * Get cluster max priority.
   * 
   * @return maximum priority of cluster
   */
  Priority getMaxClusterLevelAppPriority();

  /**
   * Get SchedulerNode corresponds to nodeId.
   *
   * @param nodeId the node id of RMNode
   *
   * @return SchedulerNode corresponds to nodeId
   */
  SchedulerNode getSchedulerNode(NodeId nodeId);

  /**
   * Normalize a resource request using scheduler level maximum resource or
   * queue based maximum resource.
   *
   * @param requestedResource the resource to be normalized
   * @param maxResourceCapability Maximum container allocation value, if null or
   *          empty scheduler level maximum container allocation value will be
   *          used
   * @return the normalized resource
   */
  Resource getNormalizedResource(Resource requestedResource,
      Resource maxResourceCapability);

  /**
   * Verify whether a submitted application lifetime is valid as per configured
   * Queue lifetime.
   * @param queueName Name of the Queue
   * @param lifetime configured application lifetime
   * @param app details of app
   * @return valid lifetime as per queue
   */
  @Public
  @Evolving
  long checkAndGetApplicationLifetime(String queueName, long lifetime,
                                      RMAppImpl app);

  /**
   * Get maximum lifetime for a queue.
   * @param queueName to get lifetime
   * @return maximum lifetime in seconds
   */
  @Public
  @Evolving
  long getMaximumApplicationLifetime(String queueName);
}

 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

<configuration>
    <!-- Define the Capacity Scheduler properties -->
    
    <!-- Enable Capacity Scheduler -->
    <property>
        <name>yarn.scheduler.capacity.enabled</name>
        <value>true</value>
    </property>

    <!-- Define the root queue capacity -->
    <property>
        <name>yarn.scheduler.capacity.root.capacity</name>
        <value>100</value>
    </property>

    <!-- Define the queues under root -->
    <property>
        <name>yarn.scheduler.capacity.root.default.capacity</name>
        <value>50</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.marketing.capacity</name>
        <value>30</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.sales.capacity</name>
        <value>20</value>
    </property>

    <!-- Define the queue capacities and ensure no queue can exceed its share -->
    <property>
        <name>yarn.scheduler.capacity.root.default.maximum-capacity</name>
        <value>50</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.marketing.maximum-capacity</name>
        <value>30</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.sales.maximum-capacity</name>
        <value>30</value>
    </property>

    <!-- Configure the minimum resource guarantees for each queue -->
    <property>
        <name>yarn.scheduler.capacity.root.default.minimum-capacity</name>
        <value>10</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.marketing.minimum-capacity</name>
        <value>10</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.sales.minimum-capacity</name>
        <value>10</value>
    </property>

    <!-- Preemption settings: enabling preemption for root queues -->
    <property>
        <name>yarn.scheduler.capacity.root.default.preemption</name>
        <value>true</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.marketing.preemption</name>
        <value>false</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.sales.preemption</name>
        <value>false</value>
    </property>

    <!-- Other optional settings: User/Queue ACLs, access control, etc. -->
    <property>
        <name>yarn.scheduler.capacity.root.default.user-limit-factor</name>
        <value>1.0</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.marketing.user-limit-factor</name>
        <value>1.0</value>
    </property>

    <property>
        <name>yarn.scheduler.capacity.root.sales.user-limit-factor</name>
        <value>1.0</value>
    </property>
    
</configuration>

 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

<configuration>
    <!-- Define the Fair Scheduler properties -->
    
    <!-- Enable Fair Scheduler -->
    <property>
        <name>yarn.scheduler.fair.enable</name>
        <value>true</value>
    </property>

    <!-- Define the root pool -->
    <property>
        <name>yarn.scheduler.fair.allocation.file</name>
        <value>/etc/hadoop/fair-scheduler.xml</value>
    </property>

    <!-- Pool configuration for different applications -->
    <property>
        <name>yarn.scheduler.fair.pool.root.capacity</name>
        <value>100</value>
    </property>

    <!-- Sub-pool under the root pool -->
    <property>
        <name>yarn.scheduler.fair.pool.root.default.capacity</name>
        <value>60</value>
    </property>

    <property>
        <name>yarn.scheduler.fair.pool.root.sales.capacity</name>
        <value>20</value>
    </property>

    <property>
        <name>yarn.scheduler.fair.pool.root.marketing.capacity</name>
        <value>20</value>
    </property>

    <!-- Ensure fair allocation between pools -->
    <property>
        <name>yarn.scheduler.fair.pool.root.default.fair-share</name>
        <value>1000</value>
    </property>

    <property>
        <name>yarn.scheduler.fair.pool.root.sales.fair-share</name>
        <value>500</value>
    </property>

    <property>
        <name>yarn.scheduler.fair.pool.root.marketing.fair-share</name>
        <value>500</value>
    </property>

    <!-- Minimum resource allocation per pool -->
    <property>
        <name>yarn.scheduler.fair.pool.root.default.minimum-share</name>
        <value>50</value>
    </property>

    <property>
        <name>yarn.scheduler.fair.pool.root.sales.minimum-share</name>
        <value>50</value>
    </property>

    <property>
        <name>yarn.scheduler.fair.pool.root.marketing.minimum-share</name>
        <value>50</value>
    </property>

    <!-- Preemption configuration (optional) -->
    <property>
        <name>yarn.scheduler.fair.pool.root.default.preemption</name>
        <value>true</value>
    </property>
    
</configuration>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

#!/bin/bash
MEMORY_RATIO=0.1
freeMem=`grep MemFree /proc/meminfo | awk '{print $2}'`
totalMem=`grep MemTotal /proc/meminfo | awk '{print $2}'`
limitMem=`echo | awk '{print int("'"$totalMem"'"*"'"$MEMORY_RATIO'"")}'`
if [ $freeMem -lt $limitMem ];then
  echo "ERROR, totalMem=$totalMem, freeMem=$freeMem, limitMem=$limitMem"
else
  echo "OK, totalMem=$totalMem, freeMem=$freeMem, limitMem=$limitMem"
fi

1

bin/yarn logs -applicationId application_1304487270789_0001

1

bin/yarn logs -applicationId application_1304487270789_0001 -containerId container_1304487270789_0001_000002 -nodeAddress 127.0.0.1_45454