src/spark_NB.py
def read_data(byte_data_directory, x_filename, y_filename=None):
    """
    reads in byte date from a list of filenames given in file located
    at x_filename. if y_filename is supplied labels will be read in and a map
    will be created as well and a label column added to the returned dataframe
    """

    X_files = sc.textFile(x_filename).collect()

    X_filenames = list(map(lambda x: byte_data_directory+x+'.bytes', X_files))
    dat = sc.wholeTextFiles(",".join(X_filenames), minPartitions=300)
    X_df = sc.parallelize(X_filenames, numSlices=300).map(lambda x: (x,x.split('/')[-1])).toDF(['filename','byte'])
    X_df = X_df.withColumn("idx", F.monotonically_increasing_id())

    if(y_filename is not None):
        y_labels = sc.textFile(y_filename).collect()
        label_map = sc.broadcast(dict(zip(X_filenames, y_labels)))
        dat = dat.map(lambda x: (x[0], x[1], int(label_map.value[x[0]])))
        dat = dat.toDF(['filename', 'text', 'label'])
    else:
        dat = dat.toDF(['filename', 'text'])
        dat.show()
    dat = X_df.join(dat, X_df.filename == dat.filename, how='left').sort("idx")
    print(dat.rdd.getNumPartitions())
    return(dat)

    reads in byte date from a list of filenames given in file located
    at x_filename. if y_filename is supplied labels will be read in and a map
    will be created as well and a label column added to the returned dataframe
    
reads in byte date from a list of filenames given in file located
at x_filename. if y_filename is supplied labels will be read in and a map
will be created as well and a label column added to the returned dataframe
reads in byte date from a list of filenames given in file located at x_filename . if y_filename is supplied labels will be read in and a map will be created as well and a label column added to the returned dataframe
def read_data(byte_data_directory, x_filename, y_filename=None):
    X_files = sc.textFile(x_filename).collect()
    X_filenames = list(map(lambda x: byte_data_directory+x+'.bytes', X_files))
    dat = sc.wholeTextFiles(",".join(X_filenames), minPartitions=300)
    X_df = sc.parallelize(X_filenames, numSlices=300).map(lambda x: (x,x.split('/')[-1])).toDF(['filename','byte'])
    X_df = X_df.withColumn("idx", F.monotonically_increasing_id())
    if(y_filename is not None):
        y_labels = sc.textFile(y_filename).collect()
        label_map = sc.broadcast(dict(zip(X_filenames, y_labels)))
        dat = dat.map(lambda x: (x[0], x[1], int(label_map.value[x[0]])))
        dat = dat.toDF(['filename', 'text', 'label'])
    else:
        dat = dat.toDF(['filename', 'text'])
        dat.show()
    dat = X_df.join(dat, X_df.filename == dat.filename, how='left').sort("idx")
    print(dat.rdd.getNumPartitions())
    return(dat)
def read_data ( byte_data_directory , x_filename , y_filename = None ) : X_files = sc . textFile ( x_filename ) . collect ( ) X_filenames = list ( map ( lambda x : byte_data_directory + x + '.bytes' , X_files ) ) dat = sc . wholeTextFiles ( "," . join ( X_filenames ) , minPartitions = 300 ) X_df = sc . parallelize ( X_filenames , numSlices = 300 ) . map ( lambda x : ( x , x . split ( '/' ) [ - 1 ] ) ) . toDF ( [ 'filename' , 'byte' ] ) X_df = X_df . withColumn ( "idx" , F . monotonically_increasing_id ( ) ) if ( y_filename is not None ) : y_labels = sc . textFile ( y_filename ) . collect ( ) label_map = sc . broadcast ( dict ( zip ( X_filenames , y_labels ) ) ) dat = dat . map ( lambda x : ( x [ 0 ] , x [ 1 ] , int ( label_map . value [ x [ 0 ] ] ) ) ) dat = dat . toDF ( [ 'filename' , 'text' , 'label' ] ) else : dat = dat . toDF ( [ 'filename' , 'text' ] ) dat . show ( ) dat = X_df . join ( dat , X_df . filename == dat . filename , how = 'left' ) . sort ( "idx" ) print ( dat . rdd . getNumPartitions ( ) ) return ( dat )
reads in byte date from a list of filenames given in file located
at x_filename.
reads in byte date from a list of filenames given in file located at x_filename .
"""
    reads in byte date from a list of filenames given in file located
    at x_filename. if y_filename is supplied labels will be read in and a map
    will be created as well and a label column added to the returned dataframe
    """
byte_data_directory
x_filename
y_filename