# -*- coding: utf-8 -*- __author__ = 'chunk' from ..common import * from .dependencies import * from . import * import rdd from .rdd import * import sys from pyspark import RDD from pyspark import SparkConf, SparkContext from pyspark.mllib.classification import LogisticRegressionWithSGD, SVMWithSGD from pyspark.mllib.regression import LabeledPoint import numpy as np np.random.seed(sum(map(ord, "whoami"))) package_dir = os.path.dirname(os.path.abspath(__file__)) class Sparker(object): def __init__(self, host='HPC-server', appname='NewPySparkApp', **kwargs): load_env() self.host = host self.appname = appname self.master = kwargs.get('master', 'spark://%s:7077' % self.host) self.conf = SparkConf() self.conf.setSparkHome(self.host) \ .setMaster(self.master) \ .setAppName(self.appname) # self.conf.set("spark.akka.frameSize","10685760") # self.conf.set("spark.driver.extraClassPath", extraClassPath) \ # .set("spark.executor.extraClassPath", extraClassPath) \ # .set("SPARK_CLASSPATH", extraClassPath) \ # .set("spark.driver.memory", "1G") \ # .set("spark.yarn.jar", sparkJar) self.sc = SparkContext(conf=self.conf) self.model = None def read_hbase(self, table_name, func=None, collect=False, parallelism=30): """ ref - http://happybase.readthedocs.org/en/latest/user.html#retrieving-data Filter format: columns=['cf1:col1', 'cf1:col2'] or columns=['cf1'] """ hconf = { "hbase.zookeeper.quorum": "HPC-server, HPC, HPC2", # "hbase.zookeeper.quorum": self.host, "hbase.mapreduce.inputtable": table_name, } hbase_rdd = self.sc.newAPIHadoopRDD(inputFormatClass=hparams["inputFormatClass"], keyClass=hparams["readKeyClass"], valueClass=hparams["readValueClass"], keyConverter=hparams["readKeyConverter"], valueConverter=hparams["readValueConverter"], conf=hconf) parser = func if func != None else rddparse_data_CV hbase_rdd = hbase_rdd.map(lambda x: parser(x)) if collect: return hbase_rdd.collect() else: """ RDD-hbase bug fixed.(with 'repartition()') When Spark reads a file from HDFS, it creates a single partition for a single input split. Input split is set by the Hadoop InputFormat used to read this file. For instance, if you use textFile() it would be TextInputFormat in Hadoop, which would return you a single partition for a single block of HDFS (but the split between partitions would be done on line split, not the exact block split), unless you have a compressed text file. In case of compressed file you would get a single partition for a single file (as compressed text files are not splittable). When you call rdd.repartition(x) it would perform a shuffle of the data from N partititons you have in rdd to x partitions you want to have, partitioning would be done on round robin basis. If you have a 30GB uncompressed text file stored on HDFS, then with the default HDFS block size setting (128MB) it would be stored in 235 blocks, which means that the RDD you read from this file would have 235 partitions. When you call repartition(1000) your RDD would be marked as to be repartitioned, but in fact it would be shuffled to 1000 partitions only when you will execute an action on top of this RDD (lazy execution concept) """ return hbase_rdd.repartition(parallelism) def write_hbase(self, table_name, data, fromrdd=False, columns=None, withdata=False): """ Data Format: (Deprecated) e.g. [["row8", "f1", "", "caocao cao"], ["row9", "f1", "c1", "asdfg hhhh"]] Data(from dictionary): e.g. data ={'row1':[1,3400,'hello'], 'row2':[34,5000,'here in mine']}, cols = ['cf_info:id', 'cf_info:size', 'cf_tag:desc'] Data(from Rdd): e.g. data =[('row1',[1,3400,'hello']), ('row2',[34,5000,'here in mine'])], cols = ['cf_info:id', 'cf_info:size', 'cf_tag:desc'] """ hconf = { "hbase.zookeeper.quorum": "HPC-server, HPC, HPC2", # "hbase.zookeeper.quorum": self.host, "hbase.mapreduce.inputtable": table_name, "hbase.mapred.outputtable": table_name, "mapreduce.outputformat.class": hparams["outputFormatClass"], "mapreduce.job.output.key.class": hparams["writeKeyClass"], "mapreduce.job.output.value.class": hparams["writeValueClass"], } cols = [col.split(':') for col in columns] if not fromrdd: rdd_data = self.sc.parallelize(data) else: rdd_data = data rdd_data.flatMap( lambda x: rdd.format_out(x, cols, withdata=withdata)).saveAsNewAPIHadoopDataset( conf=hconf, keyConverter=hparams["writeKeyConverter"], valueConverter=hparams["writeValueConverter"]) def train_svm(self, X, Y=None): if Y == None: # From rdd_labeled assert isinstance(X, RDD) svm = SVMWithSGD.train(X) else: # data = [] # for feat, tag in zip(X, Y): # data.append(LabeledPoint(tag, feat)) # svm = SVMWithSGD.train(self.sc.parallelize(data)) hdd_data = self.sc.parallelize(zip(X, Y), 30).map(lambda x: LabeledPoint(x[1], x[0])) svm = SVMWithSGD.train(hdd_data) self.model = svm # with open('res/svm_spark.model', 'wb') as modelfile: # model = pickle.dump(svm, modelfile) return self.model def predict_svm(self, x, collect=False, model=None): """ From pyspark.mlib.classification.py: >> svm.predict([1.0]) 1 >> svm.predict(sc.parallelize([[1.0]])).collect() [1] >> svm.clearThreshold() >> svm.predict(array([1.0])) 1.25... """ if model is None: if self.model != None: model = self.model else: # with open('res/svm_spark.model', 'rb') as modelfile: # model = pickle.load(modelfile) raise Exception("No model available!") res = model.predict(x) if collect: return res.collect() else: return res def test_svm(self, X, Y=None, model=None): if model is None: if self.model != None: model = self.model else: # with open('res/svm_spark.model', 'rb') as modelfile: # model = pickle.load(modelfile) raise Exception("No model available!") if Y == None: assert isinstance(X, RDD) pass else: result_Y = np.array(self.predict_svm(X, collect=True)) return np.mean(Y == result_Y)