def fit(self, Z):
<EOL>self._reset()<EOL>X = Z[:, '<STR_LIT:X>'] if isinstance(Z, DictRDD) else Z<EOL>check_rdd(X, (np.ndarray, sp.spmatrix))<EOL>def mapper(X):<EOL><INDENT>"""<STR_LIT>"""<EOL>X = check_array(X, ('<STR_LIT>', '<STR_LIT>'), dtype=np.float64)<EOL>if hasattr(X, "<STR_LIT>"):   <EOL><INDENT>mean, var = mean_variance_axis(X, axis=<NUM_LIT:0>)<EOL><DEDENT>else:<EOL><INDENT>mean, var = np.mean(X, axis=<NUM_LIT:0>), np.var(X, axis=<NUM_LIT:0>)<EOL><DEDENT>return X.shape[<NUM_LIT:0>], mean, var<EOL><DEDENT>def reducer(a, b):<EOL><INDENT>"""<STR_LIT>"""<EOL>n_a, mean_a, var_a = a<EOL>n_b, mean_b, var_b = b<EOL>n_ab = n_a + n_b<EOL>mean_ab = ((mean_a * n_a) + (mean_b * n_b)) / n_ab<EOL>var_ab = (((n_a * var_a) + (n_b * var_b)) / n_ab) +((n_a * n_b) * ((mean_b - mean_a) / n_ab) ** <NUM_LIT:2>)<EOL>return (n_ab, mean_ab, var_ab)<EOL><DEDENT>if check_rdd_dtype(X, (sp.spmatrix)):<EOL><INDENT>if self.with_mean:<EOL><INDENT>raise ValueError(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>")<EOL><DEDENT><DEDENT>self.n_samples_seen_, self.mean_, self.var_ = X.map(mapper).treeReduce(reducer)<EOL>if self.with_std:<EOL><INDENT>self.scale_ = _handle_zeros_in_scale(np.sqrt(self.var_))<EOL><DEDENT>else:<EOL><INDENT>self.scale_ = None<EOL><DEDENT>return self<EOL>
Compute the mean and std to be used for later scaling.
        Parameters
        ----------
        Z : DictRDD containing (X, y) pairs
            X - Training vector.
                {array-like, sparse matrix}, shape [n_samples, n_features]
                The data used to compute the mean and standard deviation
                used for later scaling along the features axis.
            y - Target labels
                Passthrough for ``Pipeline`` compatibility.