(ns, sc, log_details, pjob_id, module)
Pass control to the module.main method.  If module.main specifies a
    `textFile` parameter, pass the textFile instance.  Otherwise, just map
    module.main on the RDD
Pass control to the module.main method.  If module.main specifies a
    `textFile` parameter, pass the textFile instance.  Otherwise, just map
    module.main on the RDD
Pass control to the module . main method . If module . main specifies a textFile parameter pass the textFile instance . Otherwise just map module . main on the RDD
def apply_data_transform(ns, sc, log_details, pjob_id, module):
    """Pass control to the module.main method.  If module.main specifies a
    `textFile` parameter, pass the textFile instance.  Otherwise, just map
    module.main on the RDD
    """
    func_args = inspect.getargspec(module.main).args
    if 'sc' in func_args:
        log.info(
            'passing spark context to a module.main function',
            extra=log_details)
        try:
            module.main(sc=sc, ns=ns, **pjob_id)
        except Exception as err:
            log_and_raise(
                "Job failed with error: %s" % err, log_details)
    else:
        read_fp = format_fp(ns.read_fp, ns, pjob_id)
        log_details = dict(read_fp=read_fp, **log_details)
        tf = sc.textFile(read_fp, ns.minPartitions)
        tf = pre_process_data(ns=ns, tf=tf, log_details=log_details)
        if 'textFile' in func_args:
            log.info(
                'passing textFile instance to a module.main function',
                extra=log_details)
            try:
                module.main(textFile=tf, ns=ns, **pjob_id)
            except Exception as err:
                log_and_raise(
                    "Job failed with error: %s" % err, log_details)

        else:
            write_fp = format_fp(ns.write_fp, ns, pjob_id)
            log.info(
                'mapping a module.main function to all elements in a textFile'
                ' and writing output',
                extra=dict(write_fp=write_fp, **log_details))
            try:
                (
                    tf
                    .map(functools.partial(module.main, ns=ns, **pjob_id))
                    .saveAsTextFile(write_fp)
                )
            except Exception as err:
                log_and_raise(err, log_details)
def apply_data_transform ( ns , sc , log_details , pjob_id , module ) : func_args = inspect . getargspec ( module . main ) . args if 'sc' in func_args : log . info ( 'passing spark context to a module.main function' , extra = log_details ) try : module . main ( sc = sc , ns = ns , * * pjob_id ) except Exception as err : log_and_raise ( "Job failed with error: %s" % err , log_details ) else : read_fp = format_fp ( ns . read_fp , ns , pjob_id ) log_details = dict ( read_fp = read_fp , * * log_details ) tf = sc . textFile ( read_fp , ns . minPartitions ) tf = pre_process_data ( ns = ns , tf = tf , log_details = log_details ) if 'textFile' in func_args : log . info ( 'passing textFile instance to a module.main function' , extra = log_details ) try : module . main ( textFile = tf , ns = ns , * * pjob_id ) except Exception as err : log_and_raise ( "Job failed with error: %s" % err , log_details ) else : write_fp = format_fp ( ns . write_fp , ns , pjob_id ) log . info ( 'mapping a module.main function to all elements in a textFile' ' and writing output' , extra = dict ( write_fp = write_fp , * * log_details ) ) try : ( tf . map ( functools . partial ( module . main , ns = ns , * * pjob_id ) ) . saveAsTextFile ( write_fp ) ) except Exception as err : log_and_raise ( err , log_details )