<commit_before>#!/usr/bin/python2
# Flictionary Learning: Rank-1 Dictionary Learning in Flink!
# Probably PEP 8 compliant.

import argparse
import itertools
import random
import datetime
import os
import math
import sys
from flink.functions.Aggregation import Sum
from flink.functions.FlatMapFunction import FlatMapFunction
from flink.functions.GroupReduceFunction import GroupReduceFunction
from flink.plan.Constants import Order, WriteMode
from flink.plan.Environment import get_environment


###################################
# Flink helper functions
###################################

def parse_and_normalize(line):
    """
    Utility function. Parses a line of text into a floating point array, then
    whitens the array.
    """
    x = tuple(map(float, line.strip().split()))

    # x.strip() -- strips off trailing whitespace from the string
    # .split("\t") -- splits the string into a list of strings, splitting on tabs
    # map(float, list) -- converts each element of the list from strings to floats
    # np.array(list) -- converts the list of floats into a numpy array

    # comment by Xiang: the following normalization commands work for vector u,
    # but not work here. I have double-checked it with pre-normalized matrix;
    # x -= x.mean()  # 0-mean.
    # x /= sla.norm(x)  # Unit norm.
    return x


class SExploderGroupReducer(GroupReduceFunction):
    def reduce(self, iterator, collector):
        for x in iterator:
            data = enumerate(parse_and_normalize(x[1]))
            for y in data:
                collector.collect((x[0], y[0], y[1]))


def get_top_v(r, u, s):
    """
    Calculates v, and then returns the top R elements of v
    :param r: R
    :param u: Old u
    :param s: S matrix
    :return: Top R elements of v
    """
    v_ = vector_matrix(u, s)

    # sort v by using sort_group after grouping on a dummy field
    v_ = v_.map(lambda x: (x[0], x[1], 0)).name('VPreSorter')
    v_ = v_ \
        .group_by(2).sort_group(1, Order.DESCENDING) \
        .reduce_group(lambda i, c: [(i_[0], i_[1]) for i_ in i]) \
        .name('VSorter')

    v_top_R_ = v_.first(r).name('VFirst')
    return v_top_R_


def vector_matrix(u__, S_):
    """
    Generates v using u.
    """
    v_ = u__ \
        .join(S_).where(0).equal_to(0) \
        .using(lambda u_el, s_el: (s_el[1], s_el[2] * u_el[1])) \
        .name('VectorMatrix')

    v_ = v_ \
        .group_by(0) \
        .aggregate(Sum, 1) \
        .name('VectorMatrixPost')

    return v_


class RandomVectorFlatMapper(FlatMapFunction):
    """
    Takes a DataSet and replaces a single placeholder element with random numbers in [0.0, 1.0).
    Does NOT normalize.
    """
    def __init__(self, t_):
        self.T = t_
        super(RandomVectorFlatMapper, self).__init__()

    def flat_map(self, value, collector):
        vec = [random.random() for i in range(self.T)]
        return list(enumerate(vec))


class NormalizeVectorGroupReducer(GroupReduceFunction):
    """
    Normalizes a vector in (index, value) format.
    """
    def reduce(self, iterator, collector):
        data = list(iterator)
        mean = 0
        mag = 0
        length = len(data)

        for val in data:
            mean += val[1]
            mag += (val[1]) ** 2

        mean /= length
        mag = math.sqrt(mag)

        for val in data:
            new_val = val[1] - mean
            new_val /= mag
            collector.collect((val[0], new_val))


class MagnitudeGroupReducer(GroupReduceFunction):
    """
    Calculates the magnitude of a vector.
    """
    def reduce(self, iterator, collector):
        mag = 0
        for val in iterator:
            mag += (val[1]) ** 2
        mag = math.sqrt(mag)
        collector.collect((0, mag))


if __name__ == "__main__":

    print datetime.datetime.now()

    parser = argparse.ArgumentParser(description='Flictionary Learning', add_help='How to use',
                                     prog='.../pyflink2.sh R1DL_Flink.py -')

    parser.add_argument("-l", "--local", action='store_true',
                        help="Run script on the local machine i.e. NOT a cluster. (optional)")

    # Inputs.
    parser.add_argument("-i", "--input", required=True,
                        help="Input file containing the matrix S.")
    parser.add_argument("-t", "--rows", type=int, required=True,
                        help="Number of rows (observations) in the input matrix S.")
    parser.add_argument("-p", "--cols", type=int, required=True,
                        help="Number of columns (features) in the input matrix S.")
    parser.add_argument("-r", "--pnonzero", type=float, required=True,<commit_msg>Optimize imports, remove time printing (Flink handles timestamps)<commit_after>#!/usr/bin/python2
# Flictionary Learning: Rank-1 Dictionary Learning in Flink!
# Probably PEP 8 compliant.

import argparse
import math
import os
import random
import sys

from flink.functions.Aggregation import Sum
from flink.functions.FlatMapFunction import FlatMapFunction
from flink.functions.GroupReduceFunction import GroupReduceFunction
from flink.plan.Constants import Order, WriteMode
from flink.plan.Environment import get_environment


###################################
# Flink helper functions
###################################

def parse_and_normalize(line):
    """
    Utility function. Parses a line of text into a floating point array, then
    whitens the array.
    """
    x = tuple(map(float, line.strip().split()))

    # x.strip() -- strips off trailing whitespace from the string
    # .split("\t") -- splits the string into a list of strings, splitting on tabs
    # map(float, list) -- converts each element of the list from strings to floats
    # np.array(list) -- converts the list of floats into a numpy array

    # comment by Xiang: the following normalization commands work for vector u,
    # but not work here. I have double-checked it with pre-normalized matrix;
    # x -= x.mean()  # 0-mean.
    # x /= sla.norm(x)  # Unit norm.
    return x


class SExploderGroupReducer(GroupReduceFunction):
    def reduce(self, iterator, collector):
        for x in iterator:
            data = enumerate(parse_and_normalize(x[1]))
            for y in data:
                collector.collect((x[0], y[0], y[1]))


def get_top_v(r, u, s):
    """
    Calculates v, and then returns the top R elements of v
    :param r: R
    :param u: Old u
    :param s: S matrix
    :return: Top R elements of v
    """
    v_ = vector_matrix(u, s)

    # sort v by using sort_group after grouping on a dummy field
    v_ = v_.map(lambda x: (x[0], x[1], 0)).name('VPreSorter')
    v_ = v_ \
        .group_by(2).sort_group(1, Order.DESCENDING) \
        .reduce_group(lambda i, c: [(i_[0], i_[1]) for i_ in i]) \
        .name('VSorter')

    v_top_R_ = v_.first(r).name('VFirst')
    return v_top_R_


def vector_matrix(u__, S_):
    """
    Generates v using u.
    """
    v_ = u__ \
        .join(S_).where(0).equal_to(0) \
        .using(lambda u_el, s_el: (s_el[1], s_el[2] * u_el[1])) \
        .name('VectorMatrix')

    v_ = v_ \
        .group_by(0) \
        .aggregate(Sum, 1) \
        .name('VectorMatrixPost')

    return v_


class RandomVectorFlatMapper(FlatMapFunction):
    """
    Takes a DataSet and replaces a single placeholder element with random numbers in [0.0, 1.0).
    Does NOT normalize.
    """
    def __init__(self, t_):
        self.T = t_
        super(RandomVectorFlatMapper, self).__init__()

    def flat_map(self, value, collector):
        vec = [random.random() for i in range(self.T)]
        return list(enumerate(vec))


class NormalizeVectorGroupReducer(GroupReduceFunction):
    """
    Normalizes a vector in (index, value) format.
    """
    def reduce(self, iterator, collector):
        data = list(iterator)
        mean = 0
        mag = 0
        length = len(data)

        for val in data:
            mean += val[1]
            mag += (val[1]) ** 2

        mean /= length
        mag = math.sqrt(mag)

        for val in data:
            new_val = val[1] - mean
            new_val /= mag
            collector.collect((val[0], new_val))


class MagnitudeGroupReducer(GroupReduceFunction):
    """
    Calculates the magnitude of a vector.
    """
    def reduce(self, iterator, collector):
        mag = 0
        for val in iterator:
            mag += (val[1]) ** 2
        mag = math.sqrt(mag)
        collector.collect((0, mag))


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Flictionary Learning', add_help='How to use',
                                     prog='.../pyflink2.sh R1DL_Flink.py -')

    parser.add_argument("-l", "--local", action='store_true',
                        help="Run script on the local machine i.e. NOT a cluster. (optional)")

    # Inputs.
    parser.add_argument("-i", "--input", required=True,
                        help="Input file containing the matrix S.")
    parser.add_argument("-t", "--rows", type=int, required=True,
                        help="Number of rows (observations) in the input matrix S.")
    parser.add_argument("-p", "--cols", type=int, required=True,
                        help="Number of columns (features) in the input matrix S.")
    parser.add_argument("-r", "--pnonzero", type=float, required=True,
{'fraction_non_alphanumeric': 0.1330117899249732, 'fraction_numerical': 0.008360128617363344, 'mean_word_length': 2.4856182293612252, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '859526', 'n_tokens_mistral': 2945, 'n_tokens_neox': 2827, 'n_words': 868}