docs: def main():
      """Runs test imaging pipeline using Spark."""
      # Check command line arguments.
      if len(sys.argv) < 3:
          raise RuntimeError(
              'Usage: spark-submit spark_imager_test.py <settings_file> <dir> '
              '[partitions]')
  
      # Create log object.
      log = logging.getLogger('pyspark')
      log.setLevel(logging.INFO)
      log.addHandler(logging.StreamHandler(sys.stdout))
  
      # Load pipeline settings.
      with open(sys.argv[1]) as f:
          settings = json.load(f)
  
      # Get a list of input Measurement Sets to process.
      data_dir = str(sys.argv[2])
      inputs = glob(join(data_dir, '*.ms')) + glob(join(data_dir, '*.MS'))
      inputs = filter(None, inputs)
      log.info('Found input Measurement Sets: %s', ', '.join(inputs))
  
      # Get a Spark context.
      context = pyspark.SparkContext(appName="spark_imager_test")
  
      # Create the Spark RDD containing the input filenames,
      # suitably parallelized.
      partitions = int(sys.argv[3]) if len(sys.argv) > 3 else 2
      rdd = context.parallelize(inputs, partitions)
  
      # Define Spark broadcast variables.
      bc_settings = context.broadcast(settings)
      bc_grid_weights = None
  
      # Process coordinates first if required.
      if (settings['combine'] and (
              settings['imager']['weighting'] == 'Uniform' or
              settings['imager']['algorithm'] == 'W-projection')):
          # Create RDD to generate weights grids.
          rdd_coords = rdd.map(
              partial(node_run, coords_only=True, bc_settings=bc_settings,
                      bc_grid_weights=None))
  
          # Mark this RDD as persistent so it isn't computed more than once.
          rdd_coords.persist()
  
          # Get the maximum number of W-planes required, and update settings.
          num_w_planes = rdd_coords.map(lambda x: x[1]).max()
          settings['imager']['num_w_planes'] = num_w_planes
  
          # Get the combined grid of weights and broadcast it to nodes.
          output = rdd_coords.reduce(reduce_sequences)
          bc_grid_weights = context.broadcast(output[0])
  
          # Delete this RDD.
          rdd_coords.unpersist()
  
      # Re-broadcast updated settings.
      bc_settings = context.broadcast(settings)
  
      # Run parallel pipeline on worker nodes and combine visibility grids.
      output = rdd.map(
          partial(node_run, coords_only=False, bc_settings=bc_settings,
                  bc_grid_weights=bc_grid_weights)).reduce(reduce_sequences)
  
      # Finalise combined visibility grids if required.
      if settings['combine']:
          # Create an imager to finalise (FFT) the gridded data.
          imager = oskar.Imager(settings['precision'])
          for key, value in settings['imager'].items():
              setattr(imager, key, value)
  
          # Finalise grid and save image.
          save_image(imager, output[0], output[1], settings['output_file'])
          log.info('Finished. Output file is %s', settings['output_file'])
  
      context.stop()