Use trained sklearn model with pyspark

 

from pyspark import SparkContextimport numpy as npfrom sklearn import ensembledef batch(xs):    yield list(xs)N = 1000train_x = np.random.randn(N, 10)train_y = np.random.binomial(1, 0.5, N)model = ensemble.RandomForestClassifier(10).fit(train_x, train_y)test_x = np.random.randn(N * 100, 10)sc = SparkContext()n_partitions = 10rdd = sc.parallelize(test_x, n_partitions).zipWithIndex()b_model = sc.broadcast(model)result = rdd.mapPartitions(batch) \    .map(lambda xs: ([x[0] for x in xs], [x[1] for x in xs])) \    .flatMap(lambda x: zip(x[1], b_model.value.predict(x[0])))print(result.take(100))

output:

[(0, 0), (1, 1), (2, 1), (3, 1), (4, 1), (5, 0), (6, 1), (7, 0), (8, 1), (9, 1), (10, 0), (11, 1), (12, 0), (13, 0), (14, 1), (15, 0), (16, 0), (17, 1), (18, 0), (19, 0), (20, 1), (21, 0), (22, 1), (23, 1), (24, 1), (25, 1), (26, 0), (27, 0), (28, 1), (29, 0), (30, 0), (31, 0), (32, 0), (33, 1), (34, 1), (35, 1), (36, 1), (37, 1), (38, 1), (39, 0), (40, 1), (41, 1), (42, 1), (43, 0), (44, 0), (45, 0), (46, 1), (47, 1), (48, 0), (49, 0), (50, 0), (51, 0), (52, 0), (53, 0), (54, 1), (55, 0), (56, 0), (57, 0), (58, 1), (59, 0), (60, 0), (61, 0), (62, 0), (63, 0), (64, 0), (65, 1), (66, 1), (67, 1), (68, 0), (69, 0), (70, 1), (71, 1), (72, 1), (73, 0), (74, 0), (75, 1), (76, 1), (77, 0), (78, 1), (79, 0), (80, 0), (81, 0), (82, 0), (83, 0), (84, 0), (85, 1), (86, 1), (87, 0), (88, 0), (89, 0), (90, 1), (91, 0), (92, 0), (93, 0), (94, 0), (95, 0), (96, 1), (97, 1), (98, 0), (99, 1)]

 

>>> rdd.take(3)                                                                

18/05/15 09:37:18 WARN TaskSetManager: Stage 1 contains a task of very large size (723 KB). The maximum recommended task size is 100 KB.

[(array([-0.3142169 , -1.80738243, -1.29601447, -1.42500793, -0.49338668,

        0.32582428,  0.15244227, -2.41823997, -1.51832682, -0.32027413]), 0), (array([-0.00811787,  1.1534555 ,  0.92534192,  0.27246042,  1.06946727,

       -0.1420289 ,  0.3740049 , -1.84253399,  0.55459764, -0.96438845]), 1), (array([ 1.21547425,  0.87202465,  3.00628464, -1.0732967 , -1.79575235,

       -0.71943746,  0.83692206,  1.87272991,  0.31497977, -0.84061547]), 2)]

>>rdd.mapPartitions(batch).take(3)

[...,

# one element==>

[(array([ 0.95648585,  0.15749105, -1.2850535 ,  1.10495528, -1.98184263,

       -0.11160677, -0.11004717, -0.26977669,  0.93867963,  0.28810482]), 29691),

(array([ 2.67605744,  0.3678955 , -1.10677742,  1.3090983 ,  0.33327663,

       -0.29876755, -0.00869512, -0.53998984, -2.07484434, -0.83550041]), 29692),

(array([-0.23798771, -1.43967907,  0.05633439, -0.45039489, -1.47068918,

       -2.09854387, -0.70119312, -1.93214578,  0.44166082, -0.1442232 ]), 29693),

(array([-1.21476146, -0.7558832 , -0.53902146, -0.48273363, -0.24050023,

       -1.11263081, -0.02150105,  0.20790397,  0.78268026, -1.53404034]), 29694),

(array([ -9.63973837e-01,   3.51228982e-01,   3.51805780e-01,

        -5.06041907e-01,  -2.06905036e+00,  -8.66070627e-04,

        -1.11580654e+00,   4.94298203e-01,  -2.68946627e-01,

        -9.61166626e-01]), 29695)]

]

 

ref:

https://gist.github.com/lucidfrontier45/591be3eb78557d1844ca

 

https://stackoverflow.com/questions/42887621/how-to-do-prediction-with-sklearn-model-inside-spark/42887751

Well,

I will show an example of linear regression in Sklearn and show you how to use that to predict elements in Spark RDD.

First training the model with sklearn example:

# Create linear regression objectregr = linear_model.LinearRegression() # Train the model using the training sets regr.fit(diabetes_X_train, diabetes_y_train)

Here we just have the fit, and you need to predict each data from an RDD.

Your RDD in this case should be a RDD with X like this:

rdd = sc.parallelize([1, 2, 3, 4])

So you first need to broadcast your model of sklearn:

regr_bc = self.sc.broadcast(regr)

Then you can use it to predict your data like this:

rdd.map(lambda x: (x, regr_bc.value.predict(x))).collect()

So your element in the RDD is your X and the seccond element is going to be your predicted Y. The collect will return somthing like this:

[(1, 2), (2, 4), (3, 6), ...]

Well,

I will show an example of linear regression in Sklearn and show you how to use that to predict elements in Spark RDD.

First training the model with sklearn example:

# Create linear regression objectregr = linear_model.LinearRegression() # Train the model using the training sets regr.fit(diabetes_X_train, diabetes_y_train)

Here we just have the fit, and you need to predict each data from an RDD.

Your RDD in this case should be a RDD with X like this:

rdd = sc.parallelize([1, 2, 3, 4])

So you first need to broadcast your model of sklearn:

regr_bc = self.sc.broadcast(regr)

Then you can use it to predict your data like this:

rdd.map(lambda x: (x, regr_bc.value.predict(x))).collect()

So your element in the RDD is your X and the seccond element is going to be your predicted Y. The collect will return somthing like this:

[(1, 2), (2, 4), (3, 6), ...]