If think it will be interesting to have the equivalents of mappartitions with dataframe. There are many use cases where data are processed in batch. Another example is a simple linear classifier Ax=b where A is the matrix of feature vectors, x the model and b the output. Here again the product Ax can be done efficiently for a batch of data.
I will test for the broadcast hack. But I'm wondering whether it is possible to append or zip a RDD as a new column of a Dataframe. The idea is to do mappartitions on the the RDD of the input column and then and the result as output column ? Jao > Le 3 mars 2015 à 22:04, Joseph Bradley <jos...@databricks.com> a écrit : > > I see. I think your best bet is to create the cnnModel on the master and > then serialize it to send to the workers. If it's big (1M or so), then you > can broadcast it and use the broadcast variable in the UDF. There is not a > great way to do something equivalent to mapPartitions with UDFs right now. > >> On Tue, Mar 3, 2015 at 4:36 AM, Jaonary Rabarisoa <jaon...@gmail.com> wrote: >> Here is my current implementation with current master version of spark >> >> class DeepCNNFeature extends Transformer with HasInputCol with HasOutputCol >> ... { >> >> >> override def transformSchema(...) { ... } >> >> override def transform(dataSet: DataFrame, paramMap: ParamMap): >> DataFrame = { >> >> transformSchema(dataSet.schema, paramMap, logging = true) >> val map = this.paramMap ++ paramMap >> >> val deepCNNFeature = udf((v: Vector)=> { >> val cnnModel = new CaffeModel >> cnnModel.transform(v) >> } : Vector ) >> >> >> dataSet.withColumn(map(outputCol), >> deepCNNFeature(col(map(inputCol)))) >> >> } >> } >> >> where CaffeModel is a java api to Caffe C++ model. >> >> The problem here is that for every row it will create a new instance of >> CaffeModel which is inefficient since creating a new model >> means loading a large model file. And it will transform only a single row at >> a time. Or a Caffe network can process a batch of rows efficiently. In other >> words, is it possible to create an UDF that can operatats on a partition in >> order to minimize the creation of a CaffeModel and >> to take advantage of the Caffe network batch processing ? >> >> >> >>> On Tue, Mar 3, 2015 at 7:26 AM, Joseph Bradley <jos...@databricks.com> >>> wrote: >>> I see, thanks for clarifying! >>> >>> I'd recommend following existing implementations in spark.ml transformers. >>> You'll need to define a UDF which operates on a single Row to compute the >>> value for the new column. You can then use the DataFrame DSL to create the >>> new column; the DSL provides a nice syntax for what would otherwise be a >>> SQL statement like "select ... from ...". I'm recommending looking at the >>> existing implementation (rather than stating it here) because it changes >>> between Spark 1.2 and 1.3. In 1.3, the DSL is much improved and makes it >>> easier to create a new column. >>> >>> Joseph >>> >>>> On Sun, Mar 1, 2015 at 1:26 AM, Jaonary Rabarisoa <jaon...@gmail.com> >>>> wrote: >>>> class DeepCNNFeature extends Transformer ... { >>>> >>>> override def transform(data: DataFrame, paramMap: ParamMap): DataFrame >>>> = { >>>> >>>> >>>> // How can I do a map partition on the underlying RDD and >>>> then add the column ? >>>> >>>> } >>>> } >>>> >>>>> On Sun, Mar 1, 2015 at 10:23 AM, Jaonary Rabarisoa <jaon...@gmail.com> >>>>> wrote: >>>>> Hi Joseph, >>>>> >>>>> Thank your for the tips. I understand what should I do when my data are >>>>> represented as a RDD. The thing that I can't figure out is how to do the >>>>> same thing when the data is view as a DataFrame and I need to add the >>>>> result of my pretrained model as a new column in the DataFrame. >>>>> Preciselly, I want to implement the following transformer : >>>>> >>>>> class DeepCNNFeature extends Transformer ... { >>>>> >>>>> } >>>>> >>>>>> On Sun, Mar 1, 2015 at 1:32 AM, Joseph Bradley <jos...@databricks.com> >>>>>> wrote: >>>>>> Hi Jao, >>>>>> >>>>>> You can use external tools and libraries if they can be called from your >>>>>> Spark program or script (with appropriate conversion of data types, >>>>>> etc.). The best way to apply a pre-trained model to a dataset would be >>>>>> to call the model from within a closure, e.g.: >>>>>> >>>>>> myRDD.map { myDatum => preTrainedModel.predict(myDatum) } >>>>>> >>>>>> If your data is distributed in an RDD (myRDD), then the above call will >>>>>> distribute the computation of prediction using the pre-trained model. >>>>>> It will require that all of your Spark workers be able to run the >>>>>> preTrainedModel; that may mean installing Caffe and dependencies on all >>>>>> nodes in the compute cluster. >>>>>> >>>>>> For the second question, I would modify the above call as follows: >>>>>> >>>>>> myRDD.mapPartitions { myDataOnPartition => >>>>>> val myModel = // instantiate neural network on this partition >>>>>> myDataOnPartition.map { myDatum => myModel.predict(myDatum) } >>>>>> } >>>>>> >>>>>> I hope this helps! >>>>>> Joseph >>>>>> >>>>>>> On Fri, Feb 27, 2015 at 10:27 PM, Jaonary Rabarisoa <jaon...@gmail.com> >>>>>>> wrote: >>>>>>> Dear all, >>>>>>> >>>>>>> We mainly do large scale computer vision task (image classification, >>>>>>> retrieval, ...). The pipeline is really great stuff for that. We're >>>>>>> trying to reproduce the tutorial given on that topic during the latest >>>>>>> spark summit ( >>>>>>> http://ampcamp.berkeley.edu/5/exercises/image-classification-with-pipelines.html >>>>>>> ) using the master version of spark pipeline and dataframe. The >>>>>>> tutorial shows different examples of feature extraction stages before >>>>>>> running machine learning algorithms. Even the tutorial is >>>>>>> straightforward to reproduce with this new API, we still have some >>>>>>> questions : >>>>>>> Can one use external tools (e.g via pipe) as a pipeline stage ? An >>>>>>> example of use case is to extract feature learned with convolutional >>>>>>> neural network. In our case, this corresponds to a pre-trained neural >>>>>>> network with Caffe library (http://caffe.berkeleyvision.org/) . >>>>>>> The second question is about the performance of the pipeline. Library >>>>>>> such as Caffe processes the data in batch and instancing one Caffe >>>>>>> network can be time consuming when this network is very deep. So, we >>>>>>> can gain performance if we minimize the number of Caffe network >>>>>>> creation and give data in batch to the network. In the pipeline, this >>>>>>> corresponds to run transformers that work on a partition basis and give >>>>>>> the whole partition to a single caffe network. How can we create such a >>>>>>> transformer ? >>>>>>> >>>>>>> >>>>>>> Best, >>>>>>> >>>>>>> Jao >