def test_fmod(self):
     A, A_rdd = self.make_dense_rdd((8, 3))
     B, B_rdd = self.make_dense_rdd((1, 3))
     np_res = np.fmod(A, B)
     assert_array_equal(
         A_rdd.fmod(B).toarray(), np_res
     )

 def test_remainder(self):
     A, A_rdd = self.make_dense_rdd((8, 3))
     B, B_rdd = self.make_dense_rdd((1, 3))
     np_res = np.remainder(A, B)
     assert_array_equal(
         A_rdd.remainder(B).toarray(), np_res
     )

 def test_unblocking_rdd(self):
     data = np.arange(400)
     rdd = self.sc.parallelize(data, 4)
     X = ArrayRDD(rdd, 5)
     X_unblocked = X.unblock()
     assert_is_instance(X_unblocked, RDD)
     assert_array_equal(X_unblocked.take(12), np.arange(12).tolist())

    def test_same_fit_transform(self):
        Y, Y_rdd = self.make_dense_randint_rdd(low=0, high=10, shape=(1000,))

        local = LabelEncoder()
        dist = SparkLabelEncoder()

        assert_array_equal(local.fit_transform(Y), dist.fit_transform(Y_rdd).toarray())

    def test_same_classes(self):
        Y, Y_rdd = self.make_dense_randint_rdd(low=0, high=10, shape=(1000,))

        local = LabelEncoder().fit(Y)
        dist = SparkLabelEncoder().fit(Y_rdd)

        assert_array_equal(local.classes_, dist.classes_)

 def test_true_divide(self):
     A, A_rdd = self.make_dense_rdd((8, 3))
     B, B_rdd = self.make_dense_rdd((1, 3))
     np_res = A / B
     assert_array_equal(
         A_rdd.true_divide(B).toarray(), np_res
     )

    def test_same_output(self):
        X, X_rdd = self.make_text_rdd()
        local = HashingVectorizer()
        dist = SparkHashingVectorizer()

        result_local = local.transform(X)
        result_dist = sp.vstack(dist.transform(X_rdd).collect())
        assert_array_equal(result_local.toarray(), result_dist.toarray())

    def test_transform(self):
        X, X_rdd = self.make_dense_rdd((100, 4))

        fn = lambda x: x ** 2
        X1 = list(map(fn, X_rdd.collect()))
        X2 = X_rdd.transform(fn).collect()

        assert_array_equal(X1, X2)

    def test_same_output(self):
        X, X_rdd = self.make_dict_dataset()
        local = DictVectorizer()
        dist = SparkDictVectorizer()

        result_local = local.fit_transform(X)
        result_dist = sp.vstack(dist.fit_transform(X_rdd).collect())

        assert_equal(local.vocabulary_, dist.vocabulary_)
        assert_array_equal(result_local.toarray(), result_dist.toarray())

    def test_same_output(self):
        X, X_rdd = self.make_text_rdd()
        local = CountVectorizer()
        dist = SparkCountVectorizer()

        result_local = local.fit_transform(X).toarray()
        result_dist = dist.fit_transform(X_rdd).toarray()

        assert_equal(local.vocabulary_, dist.vocabulary_)
        assert_array_equal(result_local, result_dist)

    def test_transform(self):
        data = np.arange(400).reshape((100, 4))
        rdd = self.sc.parallelize(data, 4)
        X = ArrayRDD(rdd, 5)

        fn = lambda x: x ** 2
        X1 = map(fn, X.collect())
        X2 = X.transform(fn).collect()

        assert_array_equal(X1, X2)

    def test_same_inverse_transform(self):
        Y, Y_rdd = self.make_dense_randint_rdd((1000,), low_high=(0, 10))

        local = LabelEncoder().fit(Y)
        dist = SparkLabelEncoder().fit(Y_rdd)

        assert_array_equal(
            local.inverse_transform(Y),
            dist.inverse_transform(Y_rdd).toarray()
        )

    def test_same_output_sparse(self):
        X, X_rdd = self.make_dict_dataset()
        local = DictVectorizer(sparse=True)
        dist = SparkDictVectorizer(sparse=True)

        result_local = local.fit_transform(X)
        result_dist = dist.fit_transform(X_rdd)

        assert_true(check_rdd_dtype(result_dist, (sp.spmatrix,)))
        assert_equal(local.vocabulary_, dist.vocabulary_)
        assert_array_equal(result_local.toarray(), result_dist.toarray())

    def test_convert_toarray(self):
        data = np.arange(400)
        rdd = self.sc.parallelize(data, 4)
        X = ArrayRDD(rdd, 5)
        X_array = X.toarray()
        assert_array_equal(X_array, data)

        data = [2, 3, 5, 1, 6, 7, 9, 9]
        rdd = self.sc.parallelize(data, 2)
        X = ArrayRDD(rdd)
        X_array = X.toarray()
        assert_array_equal(X_array, np.array(data))

    def test_sum(self):
        data = np.arange(400).reshape((100, 4))
        rdd = self.sc.parallelize(data)
        assert_equal(ArrayRDD(rdd).sum(), data.sum())
        assert_array_equal(ArrayRDD(rdd).sum(axis=0), data.sum(axis=0))
        assert_array_equal(ArrayRDD(rdd).sum(axis=1), data.sum(axis=1))

        data = np.arange(600).reshape((100, 3, 2))
        rdd = self.sc.parallelize(data)
        assert_equal(ArrayRDD(rdd).sum(), data.sum())
        assert_array_equal(ArrayRDD(rdd).sum(axis=0), data.sum(axis=0))
        assert_array_equal(ArrayRDD(rdd).sum(axis=1), data.sum(axis=1))
        assert_array_equal(ArrayRDD(rdd).sum(axis=2), data.sum(axis=2))

    def test_same_fit_transforms(self):
        X, X_rdd = self.make_dense_rdd((1e3, 12))

        n_components = 4
        random_state = 42
        tol = 1e-7
        local = TruncatedSVD(n_components, n_iter=5, tol=tol,
                             random_state=random_state)
        dist = SparkTruncatedSVD(n_components, n_iter=50, tol=tol,
                                 random_state=random_state)

        Z_local = local.fit_transform(X)
        Z_dist = dist.fit_transform(X_rdd).toarray()

        tol = 1e-1
        assert_array_equal(Z_local.shape, Z_dist.shape)
        assert(np.allclose(+Z_dist[:, 0], Z_local[:, 0], atol=tol) |
               np.allclose(-Z_dist[:, 0], Z_local[:, 0], atol=tol))

    def test_dummy_analyzer(self):
        X, X_rdd = self.make_text_rdd()

        def splitter(x):
            return x.split()
        X = map(splitter, X)
        X_rdd = X_rdd.map(lambda x: map(splitter, x))

        local = HashingVectorizer(analyzer=lambda x: x)
        dist = SparkHashingVectorizer(analyzer=lambda x: x)

        result_local = local.transform(X)
        result_dist = sp.vstack(dist.transform(X_rdd).collect())
        assert_array_equal(result_local.toarray(), result_dist.toarray())

        result_local = local.fit_transform(X)
        result_dist = sp.vstack(dist.fit_transform(X_rdd).collect())
        assert_array_equal(result_local.toarray(), result_dist.toarray())

    def test_limit_features(self):
        X, X_rdd = self.make_text_rdd()

        params = [{'min_df': .5},
                  {'min_df': 2, 'max_df': .9},
                  {'min_df': 1, 'max_df': .6},
                  {'min_df': 2, 'max_features': 3}]

        for paramset in params:
            local = CountVectorizer(**paramset)
            dist = SparkCountVectorizer(**paramset)

            result_local = local.fit_transform(X)
            result_dist = sp.vstack(dist.fit_transform(X_rdd).collect())

            assert_equal(local.vocabulary_, dist.vocabulary_)
            assert_array_equal(result_local.toarray(), result_dist.toarray())

            result_dist = sp.vstack(dist.transform(X_rdd).collect())
            assert_array_equal(result_local.toarray(), result_dist.toarray())

    def test_get_multiple_items(self):
        x, y = np.arange(80).reshape((40, 2)), np.arange(40)
        x_rdd = self.sc.parallelize(x, 2)
        y_rdd = self.sc.parallelize(y, 2)
        z_rdd = x_rdd.zip(y_rdd)
        z = DictRDD(z_rdd, bsize=5)

        expected = [(np.arange(0, 10).reshape((5, 2)), np.arange(0, 5)),
                    (np.arange(10, 20).reshape((5, 2)), np.arange(5, 10))]
        assert_array_equal(z[:2, 1].collect(),
                           [expected[0][1], expected[1][1]])
        assert_array_equal(z[[0, 1], 0].collect(),
                           [expected[0][0], expected[1][0]])
        assert_multiple_tuples_equal(z[[0, 1], [1]].collect(),
                                     [(expected[0][1],),
                                      (expected[1][1],)])
        assert_multiple_tuples_equal(z[[0, 1], -1:].collect(),
                                     [(expected[0][1],),
                                      (expected[1][1],)])
        assert_multiple_tuples_equal(z[[1, 0], [1, 0]].collect(),
                                     [expected[1][::-1], expected[0][::-1]])

 def test_add(self):
     A, A_rdd = self.make_dense_rdd((8, 3))
     B, B_rdd = self.make_dense_rdd((1, 3))
     np_res = A + B
     assert_array_equal(
         A_rdd.add(B).toarray(), np_res
     )
     assert_array_equal((A_rdd + B).toarray(), np_res)
     A_rdd += B
     assert_array_equal(A_rdd.toarray(), np_res)