code: def launch_cluster(conn, opts, cluster_name):
      if opts.identity_file is None:
          print(
              "ERROR: Must provide an identity file (-i) for ssh connections.",
              file=stderr,
          )
          sys.exit(1)
  
      if opts.key_pair is None:
          print(
              "ERROR: Must provide a key pair name (-k) to use on instances.",
              file=stderr,
          )
          sys.exit(1)
  
      user_data_content = None
      if opts.user_data:
          with open(opts.user_data) as user_data_file:
              user_data_content = user_data_file.read()
  
      print("Setting up security groups...")
      master_group = get_or_make_group(
          conn, cluster_name + "-master", opts.vpc_id
      )
      slave_group = get_or_make_group(
          conn, cluster_name + "-slaves", opts.vpc_id
      )
      authorized_address = opts.authorized_address
      if master_group.rules == []:  # Group was just now created
          if opts.vpc_id is None:
              master_group.authorize(src_group=master_group)
              master_group.authorize(src_group=slave_group)
          else:
              master_group.authorize(
                  ip_protocol="icmp",
                  from_port=-1,
                  to_port=-1,
                  src_group=master_group,
              )
              master_group.authorize(
                  ip_protocol="tcp",
                  from_port=0,
                  to_port=65535,
                  src_group=master_group,
              )
              master_group.authorize(
                  ip_protocol="udp",
                  from_port=0,
                  to_port=65535,
                  src_group=master_group,
              )
              master_group.authorize(
                  ip_protocol="icmp",
                  from_port=-1,
                  to_port=-1,
                  src_group=slave_group,
              )
              master_group.authorize(
                  ip_protocol="tcp",
                  from_port=0,
                  to_port=65535,
                  src_group=slave_group,
              )
              master_group.authorize(
                  ip_protocol="udp",
                  from_port=0,
                  to_port=65535,
                  src_group=slave_group,
              )
          master_group.authorize("tcp", 22, 22, authorized_address)
          master_group.authorize("tcp", 8080, 8081, authorized_address)
          master_group.authorize(
              "tcp", 18080, 18080, authorized_address
          )
          master_group.authorize(
              "tcp", 19999, 19999, authorized_address
          )
          master_group.authorize(
              "tcp", 50030, 50030, authorized_address
          )
          master_group.authorize(
              "tcp", 50070, 50070, authorized_address
          )
          master_group.authorize(
              "tcp", 60070, 60070, authorized_address
          )
          master_group.authorize("tcp", 4040, 4045, authorized_address)
          # HDFS NFS gateway requires 111,2049,4242 for tcp & udp
          master_group.authorize("tcp", 111, 111, authorized_address)
          master_group.authorize("udp", 111, 111, authorized_address)
          master_group.authorize("tcp", 2049, 2049, authorized_address)
          master_group.authorize("udp", 2049, 2049, authorized_address)
          master_group.authorize("tcp", 4242, 4242, authorized_address)
          master_group.authorize("udp", 4242, 4242, authorized_address)
          # RM in YARN mode uses 8088
          master_group.authorize("tcp", 8088, 8088, authorized_address)
          if opts.ganglia:
              master_group.authorize(
                  "tcp", 5080, 5080, authorized_address
              )
      if slave_group.rules == []:  # Group was just now created
          if opts.vpc_id is None:
              slave_group.authorize(src_group=master_group)
              slave_group.authorize(src_group=slave_group)
          else:
              slave_group.authorize(
                  ip_protocol="icmp",
                  from_port=-1,
                  to_port=-1,
                  src_group=master_group,
              )
              slave_group.authorize(
                  ip_protocol="tcp",
                  from_port=0,
                  to_port=65535,
                  src_group=master_group,
              )
              slave_group.authorize(
                  ip_protocol="udp",
                  from_port=0,
                  to_port=65535,
                  src_group=master_group,
              )
              slave_group.authorize(
                  ip_protocol="icmp",
                  from_port=-1,
                  to_port=-1,
                  src_group=slave_group,
              )
              slave_group.authorize(
                  ip_protocol="tcp",
                  from_port=0,
                  to_port=65535,
                  src_group=slave_group,
              )
              slave_group.authorize(
                  ip_protocol="udp",
                  from_port=0,
                  to_port=65535,
                  src_group=slave_group,
              )
          slave_group.authorize("tcp", 22, 22, authorized_address)
          slave_group.authorize("tcp", 8080, 8081, authorized_address)
          slave_group.authorize("tcp", 50060, 50060, authorized_address)
          slave_group.authorize("tcp", 50075, 50075, authorized_address)
          slave_group.authorize("tcp", 60060, 60060, authorized_address)
          slave_group.authorize("tcp", 60075, 60075, authorized_address)
  
      # Check if instances are already running in our groups
      existing_masters, existing_slaves = get_existing_cluster(
          conn, opts, cluster_name, die_on_error=False
      )
      if existing_slaves or (
          existing_masters and not opts.use_existing_master
      ):
          print(
              "ERROR: There are already instances running in group %s or %s"
              % (master_group.name, slave_group.name),
              file=stderr,
          )
          sys.exit(1)
  
      # Figure out Spark AMI
      if opts.ami is None:
          opts.ami = get_spark_ami(opts)
  
      # we use group ids to work around https://github.com/boto/boto/issues/350
      additional_group_ids = []
      if opts.additional_security_group:
          additional_group_ids = [
              sg.id
              for sg in conn.get_all_security_groups()
              if opts.additional_security_group in (sg.name, sg.id)
          ]
      print("Launching instances...")
  
      try:
          image = conn.get_all_images(image_ids=[opts.ami])[0]
      except:
          print("Could not find AMI " + opts.ami, file=stderr)
          sys.exit(1)
  
      # Create block device mapping so that we can add EBS volumes if asked to.
      # The first drive is attached as /dev/sds, 2nd as /dev/sdt, ... /dev/sdz
      block_map = BlockDeviceMapping()
      if opts.ebs_vol_size > 0:
          for i in range(opts.ebs_vol_num):
              device = EBSBlockDeviceType()
              device.size = opts.ebs_vol_size
              device.volume_type = opts.ebs_vol_type
              device.delete_on_termination = True
              block_map["/dev/sd" + chr(ord("s") + i)] = device
  
      # AWS ignores the AMI-specified block device mapping for M3 (see SPARK-3342).
      if opts.instance_type.startswith("m3."):
          for i in range(get_num_disks(opts.instance_type)):
              dev = BlockDeviceType()
              dev.ephemeral_name = "ephemeral%d" % i
              # The first ephemeral drive is /dev/sdb.
              name = "/dev/sd" + string.ascii_letters[i + 1]
              block_map[name] = dev
  
      # Launch slaves
      if opts.spot_price is not None:
          # Launch spot instances with the requested price
          print(
              "Requesting %d slaves as spot instances with price $%.3f"
              % (opts.slaves, opts.spot_price)
          )
          zones = get_zones(conn, opts)
          num_zones = len(zones)
          i = 0
          my_req_ids = []
          for zone in zones:
              num_slaves_this_zone = get_partition(
                  opts.slaves, num_zones, i
              )
              slave_reqs = conn.request_spot_instances(
                  price=opts.spot_price,
                  image_id=opts.ami,
                  launch_group="launch-group-%s" % cluster_name,
                  placement=zone,
                  count=num_slaves_this_zone,
                  key_name=opts.key_pair,
                  security_group_ids=[slave_group.id]
                  + additional_group_ids,
                  instance_type=opts.instance_type,
                  block_device_map=block_map,
                  subnet_id=opts.subnet_id,
                  placement_group=opts.placement_group,
                  user_data=user_data_content,
                  instance_profile_name=opts.instance_profile_name,
              )
              my_req_ids += [req.id for req in slave_reqs]
              i += 1
  
          print("Waiting for spot instances to be granted...")
          try:
              while True:
                  time.sleep(10)
                  reqs = conn.get_all_spot_instance_requests()
                  id_to_req = {}
                  for r in reqs:
                      id_to_req[r.id] = r
                  active_instance_ids = []
                  for i in my_req_ids:
                      if (
                          i in id_to_req
                          and id_to_req[i].state == "active"
                      ):
                          active_instance_ids.append(
                              id_to_req[i].instance_id
                          )
                  if len(active_instance_ids) == opts.slaves:
                      print("All %d slaves granted" % opts.slaves)
                      reservations = conn.get_all_reservations(
                          active_instance_ids
                      )
                      slave_nodes = []
                      for r in reservations:
                          slave_nodes += r.instances
                      break
                  else:
                      print(
                          "%d of %d slaves granted, waiting longer"
                          % (len(active_instance_ids), opts.slaves)
                      )
          except:
              print("Canceling spot instance requests")
              conn.cancel_spot_instance_requests(my_req_ids)
              # Log a warning if any of these requests actually launched instances:
              (master_nodes, slave_nodes) = get_existing_cluster(
                  conn, opts, cluster_name, die_on_error=False
              )
              running = len(master_nodes) + len(slave_nodes)
              if running:
                  print(
                      (
                          "WARNING: %d instances are still running"
                          % running
                      ),
                      file=stderr,
                  )
              sys.exit(0)
      else:
          # Launch non-spot instances
          zones = get_zones(conn, opts)
          num_zones = len(zones)
          i = 0
          slave_nodes = []
          for zone in zones:
              num_slaves_this_zone = get_partition(
                  opts.slaves, num_zones, i
              )
              if num_slaves_this_zone > 0:
                  slave_res = image.run(
                      key_name=opts.key_pair,
                      security_group_ids=[slave_group.id]
                      + additional_group_ids,
                      instance_type=opts.instance_type,
                      placement=zone,
                      min_count=num_slaves_this_zone,
                      max_count=num_slaves_this_zone,
                      block_device_map=block_map,
                      subnet_id=opts.subnet_id,
                      placement_group=opts.placement_group,
                      user_data=user_data_content,
                      instance_initiated_shutdown_behavior=opts.instance_initiated_shutdown_behavior,
                      instance_profile_name=opts.instance_profile_name,
                  )
                  slave_nodes += slave_res.instances
                  print(
                      "Launched {s} slave{plural_s} in {z}, regid = {r}".format(
                          s=num_slaves_this_zone,
                          plural_s=(
                              "" if num_slaves_this_zone == 1 else "s"
                          ),
                          z=zone,
                          r=slave_res.id,
                      )
                  )
              i += 1
  
      # Launch or resume masters
      if existing_masters:
          print("Starting master...")
          for inst in existing_masters:
              if inst.state not in ["shutting-down", "terminated"]:
                  inst.start()
          master_nodes = existing_masters
      else:
          master_type = opts.master_instance_type
          if master_type == "":
              master_type = opts.instance_type
          if opts.zone == "all":
              opts.zone = random.choice(conn.get_all_zones()).name
          master_res = image.run(
              key_name=opts.key_pair,
              security_group_ids=[master_group.id]
              + additional_group_ids,
              instance_type=master_type,
              placement=opts.zone,
              min_count=1,
              max_count=1,
              block_device_map=block_map,
              subnet_id=opts.subnet_id,
              placement_group=opts.placement_group,
              user_data=user_data_content,
              instance_initiated_shutdown_behavior=opts.instance_initiated_shutdown_behavior,
              instance_profile_name=opts.instance_profile_name,
          )
  
          master_nodes = master_res.instances
          print(
              "Launched master in %s, regid = %s"
              % (zone, master_res.id)
          )
  
      # This wait time corresponds to SPARK-4983
      print("Waiting for AWS to propagate instance metadata...")
      time.sleep(15)
  
      # Give the instances descriptive names and set additional tags
      additional_tags = {}
      if opts.additional_tags.strip():
          additional_tags = dict(
              map(str.strip, tag.split(":", 1))
              for tag in opts.additional_tags.split(",")
          )
  
      for master in master_nodes:
          master.add_tags(
              dict(
                  additional_tags,
                  Name="{cn}-master-{iid}".format(
                      cn=cluster_name, iid=master.id
                  ),
              )
          )
  
      for slave in slave_nodes:
          slave.add_tags(
              dict(
                  additional_tags,
                  Name="{cn}-slave-{iid}".format(
                      cn=cluster_name, iid=slave.id
                  ),
              )
          )
  
      # Return all the instances
      return (master_nodes, slave_nodes)
  
imports: from __future__ import division, print_function, with_statement
  import codecs
  import hashlib
  import itertools
  import logging
  import os
  import os.path
  import pipes
  import random
  import shutil
  import string
  from stat import S_IRUSR
  import subprocess
  import sys
  import tarfile
  import tempfile
  import textwrap
  import time
  import warnings
  from datetime import datetime
  from optparse import OptionParser
  from sys import stderr
  import boto
  from boto.ec2.blockdevicemapping import BlockDeviceMapping, BlockDeviceType, EBSBlockDeviceType
  from boto import ec2
label_desc: Use image node_id 7 for calling a global function with example usage: launch_cluster(conn, opts, cluster_name) and returns: master_nodes,  slave_nodes