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()
def main ( ) : if len ( sys . argv ) < 3 : raise RuntimeError ( 'Usage: spark-submit spark_imager_test.py <settings_file> <dir> ' '[partitions]' ) log = logging . getLogger ( 'pyspark' ) log . setLevel ( logging . INFO ) log . addHandler ( logging . StreamHandler ( sys . stdout ) ) with open ( sys . argv [ 1 ] ) as f : settings = json . load ( f ) 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 ) ) context = pyspark . SparkContext ( appName = "spark_imager_test" ) partitions = int ( sys . argv [ 3 ] ) if len ( sys . argv ) > 3 else 2 rdd = context . parallelize ( inputs , partitions ) bc_settings = context . broadcast ( settings ) bc_grid_weights = None if ( settings [ 'combine' ] and ( settings [ 'imager' ] [ 'weighting' ] == 'Uniform' or settings [ 'imager' ] [ 'algorithm' ] == 'W-projection' ) ) : rdd_coords = rdd . map ( partial ( node_run , coords_only = True , bc_settings = bc_settings , bc_grid_weights = None ) ) rdd_coords . persist ( ) num_w_planes = rdd_coords . map ( lambda x : x [ 1 ] ) . max ( ) settings [ 'imager' ] [ 'num_w_planes' ] = num_w_planes output = rdd_coords . reduce ( reduce_sequences ) bc_grid_weights = context . broadcast ( output [ 0 ] ) rdd_coords . unpersist ( ) bc_settings = context . broadcast ( settings ) output = rdd . map ( partial ( node_run , coords_only = False , bc_settings = bc_settings , bc_grid_weights = bc_grid_weights ) ) . reduce ( reduce_sequences ) if settings [ 'combine' ] : imager = oskar . Imager ( settings [ 'precision' ] ) for key , value in settings [ 'imager' ] . items ( ) : setattr ( imager , key , value ) save_image ( imager , output [ 0 ] , output [ 1 ] , settings [ 'output_file' ] ) log . info ( 'Finished. Output file is %s' , settings [ 'output_file' ] ) context . stop ( )
Runs test imaging pipeline using Spark.