Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Set Tags on EMR cluster creation #2208

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Set Tags on EMR cluster creation #2208

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 9 additions & 19 deletions mrjob/emr.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1121,13 +1121,21 @@ def _create_cluster(self, persistent=False):
emr_client = self.make_emr_client()

kwargs = self._cluster_kwargs(persistent)
kwargs["Tags"] = self._build_tags()
log.debug('Calling run_job_flow(%s)' % (
', '.join('%s=%r' % (k, v)
for k, v in sorted(kwargs.items()))))
cluster_id = emr_client.run_job_flow(**kwargs)['JobFlowId']

log.info('Created new cluster %s' % cluster_id)

return cluster_id

def _build_tags(self):
"""
Build tags to be set on cluster creation
"""

# set EMR tags for the cluster
tags = dict(self._opts['tags'])

Expand All @@ -1143,25 +1151,7 @@ def _create_cluster(self, persistent=False):
tags['__mrjob_pool_hash'] = self._pool_hash()
tags['__mrjob_pool_name'] = self._opts['pool_name']

self._add_tags(tags, cluster_id)

return cluster_id

def _add_tags(self, tags, cluster_id):
"""Add tags in the dict *tags* to cluster *cluster_id*. Do nothing
if *tags* is empty or ``None``"""
if not tags:
return

tags_items = sorted(tags.items())

self.make_emr_client().add_tags(
ResourceId=cluster_id,
Tags=[dict(Key=k, Value=v) for k, v in tags_items])

log.info('Added EMR tags to cluster %s: %s' % (
cluster_id,
', '.join('%s=%s' % (tag, value) for tag, value in tags_items)))
return [dict(Key=str(k), Value='' if v is None else str(v)) for k, v in tags.items()]

# TODO: could break this into sub-methods for clarity
def _cluster_kwargs(self, persistent=False):
Expand Down