<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 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()