anchor.diff:  
   You need to implment readTraining() of [PDataSource](https://docs.prediction.io/api/current/#io.prediction.controller.PDataSource), where you can use the [PEventStore Engine API](https://docs.prediction.io/api/current/#io.prediction.data.store.PEventStore$) to read the events and create the TrainingData based on the events.
   
  -The following code is example of reading user "view" and "buy" item events, and then filtered specific type of events for future processing and return TrainingData accordingly.
  -
  -```scala
  -
  +The following code example reads user "view" and "buy" item events, filters specific type of events for future processing and returns TrainingData accordingly.
  +
  +```scala
   class DataSource(val dsp: DataSourceParams)
     extends PDataSource[TrainingData,
         EmptyEvaluationInfo, Query, EmptyActualResult] {
  
positive.code:           throw e
        }
      }
  
  ```
  
  If you have used speical events `$set/$unset/$delete` setting entity's properties, you can retrieve it with `PEventStore.aggregateProperties()`.
  
  Please see [event modeling](https://docs.prediction.io/datacollection/eventmodel/) to understand usage of special `$set/$unset/$delete` events.
  
  For example, the following code show how you could retrieve properties of the "item" entities:
  
  ```scala
  
      // create a RDD of (entityID, Item)
      val itemsRDD: RDD[(String, Item)] = PEventStore.aggregateProperties(
        appName = dsp.appName,
  
negative.code:         logger.error(s"Error when read recent events: ${e}")
          throw e
      }
  ```
  
  
  Example:
  
  - [Algorithm of E-Commerce Recommendation template](/templates/ecommercerecommendation/dase#algorithm): LEventStore.findByEntity() is used to retrieve all items seen by the user and filter them from recommendation in predict().
  
  
  # Serving
  