Flog: the functional logger

In this series of blogs, we've covered functional versions of comparers, table parsers, and just "code."

It's time to revisit an earlier functional topic: loggers (see Spy). Spy was OK but the syntax was not ideal. In particular, it required an extra set of parentheses which had to be removed when removing the log messages. So, over the last few years, I have created a new functional logger called Flog. You can find it at Flog. Take a look, in particular, at the README.md file, and also under the test directory: FlogExamples.sc as well as other spec files in the test directory.

I have also written about it on Medium.

So, what is the real difference between Spy and Flog? It's mostly down to the use of an implicit class. What's that?

Implicit classes, like the other implicit things (vals, defs, objects) are an especially powerful bit of magic. If some code requires a particular method, say, propertyA (defined for class Y) and you try to invoke it on an instance of class X which does not have such a method, the compiler is going to look for an implicit converter from X to Y. This could be declared as an implicit def name(x: X): Y, or an implicit val of type X=>Y, or as the constructor of an implicit class Y which takes exactly one parameter, an X. This is a similar mechanism that allows for additional methods on String that belong to the StringOps class, such as r for regex.

In the case of Flog, the name (not that it matters) of the implicit class is Flogger. It defines a number of methods such as def !: X which performs info-level logging for X and various compound types of X such as Iterable[X], Option[X], etc. There are similar methods for debug (!?) and trace (!??).

The real utility of the library is in the various implicits that allow for the rendering of different types and types of types. Flog can handle Iterable[X], Option[X], Map[K, V], Future[X], Try[X], as well as any case class (or other Product).

Functionalizing code

I'm not sure if functionalizing is really a word. But let's suppose it is.

Some time ago, I wrote an application to allow me to view the output of the "Results" option that Blackboard Lean (oops: Learn) provides us. Since most answers use HTML, the content of individual answers can be obscured at best. At worst, impossible.

So, I created a filter that takes the CSV file and turns it into an HTML file with a table (aspects of questions are columns, student submissions are rows). I recently upgraded it to allow me to specify a particular set of columns that I was interested in. But I was very dissatisfied when I looked at the HTML class which I had previously used to set up the output tags. This was how it looked:

/**  * Mutable class to form an HTML string  */

class HTML() {
  val content = new StringBuilder("")
  val tagStack: mutable.Stack[String] = mutable.Stack[String]()

  def tag(w: String): StringBuilder = {
    tagStack.push(w)
    content.append(s"<$w>")
  }

  def unTag: StringBuilder = content.append(s"</${tagStack.pop()}>")

  def append(w: String): StringBuilder = content.append(w)

  def close(): Unit = while (tagStack.nonEmpty) {
    unTag
  }

  override def toString: String = content.toString + "\n"

}

And this was how it was used:

def preamble(w: String): String = {
  val result = new HTML
  result.tag("head")
  result.tag("title")
  result.append(w)
  result.close()
  result.toString
}



def parseStreamIntoHTMLTable(ws: Stream[String], title: String): String = {
  val result = new HTML
  result.tag("html")
  result.append(preamble(title))
  result.tag("body")
  result.tag("table")
  ws match {
    case header #:: body =>
      result.append(parseRowIntoHTMLRow(header, header = true))
      for (w <- body) result.append(parseRowIntoHTMLRow(w))
  }
  result.close()
  result.toString
}
 

The rendering of the code here doesn't actually show that the mutable Stack class is deprecated. How could I be using a deprecated class--and for mutation too! Ugh! Well, it was a utility, not an exemplar for my students. So, it was acceptable.
 
But I decided to functionalize it. First, I needed to create a trait which had the basic behavior I needed: 


 

/**
  * Trait Tag to model an Markup Language-type document.
  */
trait Tag {

  /**
    * Method to yield the name of this Tag
    *
    * @return the name, that's to say what goes between &lt; and &gt;
    */
  def name: String

  /**
    * Method to yield the attributes of this Tag.
    *
    * @return a sequence of attributes, each of which is a String
    */
  def attributes: Seq[String]

  /**
    * Method to yield the content of this Tag.
    *
    * @return the content as a String.
    */
  def content: String

  /**
    * Method to yield the child Tags of this Tag.
    *
    * @return a Seq of Tags.
    */
  def tags: Seq[Tag]

  /**
    * Method to add a child to this Tag
    *
    * @param tag the tag to be added
    * @return a new version of this Tag with the additional tag added as a child
    */
  def :+(tag: Tag): Tag

  /**
    * Method to yield the tag names depth-first in a Seq
    *
    * @return a sequence of tag names
    */
  def \\ : Seq[String] = name +: (for (t <- tags; x <- t.\\) yield x)
}


Together with an abstract base class:



abstract class BaseTag(name: String, attributes: Seq[String], content: String, tags: Seq[Tag])(implicit rules: TagRules) extends Tag {

  override def toString: String = s"""\n${tagString()}$content$tagsString${tagString(true)}"""

  private def attributeString(close: Boolean) = if (close || attributes.isEmpty) "" else " " + attributes.mkString(" ")

  private def tagsString = if (tags.isEmpty) "" else tags mkString ""

  private def nameString(close: Boolean = false) = (if (close) "/" else "") + name

  private def tagString(close: Boolean = false) = s"<${nameString(close)}${attributeString(close)}>"
}

And a case class for HTML with its companion object:



/**
  * Case class to model an HTML document.
  * @param name the name of the tag at the root of the document.
  * @param attributes the attributes of the tag.
  * @param content the content of the tag.
  * @param tags the child tags.
  * @param rules the "rules" (currently ignored) but useful in the future to validate documents.
  */
case class HTML(name: String, attributes: Seq[String], content: String, tags: Seq[Tag])(implicit rules: TagRules) extends BaseTag(name, attributes, content, tags) {

  /**
    * Method to add a child to this Tag
    *
    * @param tag the tag to be added
    * @return a new version of this Tag with the additional tag added as a child
    */
  override def :+(tag: Tag): Tag = HTML(name, attributes, content, tags :+ tag)
}

/**
  * Companion object to HTML
  */
object HTML {

  implicit object HtmlRules extends TagRules

  def apply(name: String, attributes: Seq[String], content: String): HTML = apply(name, attributes, content, Nil)

  def apply(name: String, attributes: Seq[String]): HTML = apply(name, attributes, "")

  def apply(name: String): HTML = apply(name, Nil)

  def apply(name: String, content: String): HTML = apply(name, Nil, content)

}


Here's one of the places the tags in the document are set up:


  def parseStreamProjectionIntoHTMLTable(columns: Seq[String], wss: Stream[Seq[String]], title: String): Try[Tag] = Try {
    val table = HTML("table", Seq("""border="1"""")) :+ parseRowProjectionIntoHTMLRow(columns, header = true)
    val body = HTML("body") :+ wss.foldLeft(table)((tag, ws) => tag :+ parseRowProjectionIntoHTMLRow(ws))
    HTML("html") :+ preamble(title) :+ body
  }


Now, all we have to do is to use .toString on the document to render it for the final HTML source!

Some observations on foldLeft with reference to short-circuiting

Have you ever written code such as the following and expected it to quit accessing values of y after it hits the first true value?

    val y = Seq[Boolean](false, true, ..., false)
    val x = y.foldLeft[Boolean](false)(_ || _)

After all, the second parameter of || is call-by-name so, if it's not needed to determine the outcome of the value for x, then that should mean that the recursion of the foldLeft should stop, right?

But it doesn't in practice, and if you look at the definition of foldLeft, you can see why. There is no short-circuit defined.

I therefore created a method in the FP module of my LaScala library to be the short-circuit equivalent of foldLeft. It's called foldLeftShort and it looks like this:

  /**
    * This method performs a foldLeft with short-circuit logic
    *
    * @param xi an iterator of X objects
    * @param y0 the initial value of Y (e.g. 0 if Y is Int)
    * @param q  the quick-return function: if this evaluates to true, given the new y value, that value is immediately returned
    * @param f  the function which accumulates the result: as with foldLeft, it takes a Y and an X and yields a Y
    * @tparam X the underlying type of the iterator
    * @tparam Y the return type
    * @return the aggregate of the given iterator, according to the initial value of y0 and the accumulator function f.
    */
  def foldLeftShort[X, Y](xi: Iterator[X], y0: Y, q: Y => Boolean)(f: (Y, X) => Y): Y = {
    def inner(_y: Y, xo: => Option[X]): Y =
      xo match {
        case None => _y
        case Some(x) =>
          val y = f(_y, x)
          if (q(y)) y
          else
            inner(y, nextOption(xi))
      }

    inner(y0, nextOption(xi))
  }

  /**
    * Method to get the next element from an iterator, as an Option.
    *
    * @param xi the iterator.
    * @tparam X the underlying type.
    * @return Some(x) where x is the next element; or None where the iterator is exhausted.
    */
  def nextOption[X](xi: Iterator[X]): Option[X] = if (xi.hasNext) Some(xi.next) else None

As with foldLeft, the accumulating function is in its own parameter set. But the initial value is joined by the iterator itself and the short-circuit function in the first parameter set. It would be nice to be able to extend the Iterator type to include this.

The inner function isn't defined in the typical way, using the concatenation unapply method (::) because, for some reason that I don't quite understand, you cannot decompose an Iterator this way. Instead, we make use of another FP method: nextOption which returns the next element of the iterator as an Option (see above).

Note the short-circuit mechanism, which really does work. As soon as we have a value by applying f to the current accumulator and the newest value from the iterator, we return if f evaluates to true.

In order to use this as efficiently, as possible, we need to write code something like the following where in this example, we want to save doing a trivial lookup too many times. That's not worth the effort--but just suppose that the lookup was actually an expensive database query instead:

    val lookupTable = Map("a"->false, "b"->true)
    val boi = ns.view.map(x => lookupTable.get(x)).toIterator
    FP.foldLeftShort[Option[Boolean], Boolean](boi, false, b => b)(_ || _.getOrElse(false))

Obviously, if you want to end up with an Option[Boolean], you would write something like this instead:

FP.foldLeftShort[Option[Boolean], Option[Boolean]](boi, Some(false), b => b.getOrElse(false))(optionalOr)

/**
    * NOTE: that we can't use map2 here because we are essentially implementing three-valued (Kleenean) logic
  *
  * @param b1 the accumulator: if b2 is None or Some(false) then b1 is returned unchanged
  * @param b2 the value: if b2 is Some(true) then Some(true) will be returned
  * @return an Optional[Boolean]
  */
def optionalOr(b1: Option[Boolean], b2: Option[Boolean]): Option[Boolean] =
    b2 match {
      case Some(x) => for (b <- b1) yield b || x
      case None => b1
    }

Note that, if you use map2(_ || _) instead of optionalOr, the logic is probably not going to give you exactly what you want, because None will combine with false to give None, whereas you really want None to have no effect on the accumulator value at all.

Scala's pesky "contravariant position" problem

I'm sure this has happened to you. You design a data structure, for example a tree and, because you are thinking essentially of using this tree to look things up (perhaps it's an AST for example) you decide to make the tree covariant in its underlying type:

trait Tree[+A] extends Node[A] {
   def children: Seq[Node[A]]
}

Now, it's time to start building this tree up from nodes and branches so you add a method like the following:

  def +:(a: A): Tree[A] = :+(a)

Aargh! it's the dreaded "Covariant type A appears in contravariant position in type A of value a." Does this mean that we have to re-write our Tree trait so that it is invariant (as we would have to do if tree was a mutable data structure like Array)? Not at all.

There is a simple and pretty much automatic way of refactoring the +: method. Use the fact that you can define a parametric type at the method level and constrain it to be a super-type of A. Thus:

def +:[B >: A](a: B): Tree[B] = :+(a)

Since B is now a super-type of A (or, of course, A itself), then we declare a (we could rename it b now of course) as a B and we return a Tree[B]. Problem solved.

Note also that if the type of a needs to have one or more context bounds you can declare these too, just as you would expect:

  def +:[B >: A : TreeBuilder](b: B): Tree[B] = :+(b)

I should perhaps point out that this problem is not unique to Scala. It will occur (or should occur) in any object-oriented language that allows generic typing (or anywhere the Liskov substitution principle applies). It's only because Java, for example, glosses over the whole concept of variance, that we don't find the problem there too.

Next time I will talk about having more than one context bound for a type.

Spying -- or a functional way to log in Scala

All of the logging utilities I've found for Scala suffer from the same problem: they are not written in a functional-programming style. Here's the problem: you want to log the result of a function which, to choose an example more or less at random, looks like this:

  def map2[T, U](to1: Option[T], to2: => Option[T])(f: (T, T) => U): Option[U] = for {t1 <- to1; t2 <- to2} yield f(t1, t2)

Now, we decide that we'd like to log the result of this method (or maybe we just want to print it to the console). To do this the "traditional" way, we would need to write something like this:

  def map2[T, U](to1: Option[T], to2: => Option[T])(f: (T, T) => U): Option[U] = {
    val logger = LoggerFactory.getLogger(getClass)
    val r = for {t1 <- to1; t2 <- to2} yield f(t1, t2)
    logger.debug(s"map2: ${r.toString}")
    r
  }

We have interrupted the flow of the expression, we've had to create a new variable (r), we've had to wrap it all in braces. Not cool!

So, I looked around for something that could do this the functional way. Unfortunately, I couldn't find anything. There is something in Scalaz but it looked complicated and maybe a little too general. So I decided (as I often do) to write my own and call it Spy.

One of the design decisions that I had to make was this: I don't want the Spy to have to be instantiated for every user invocation, or even every class that an invocation appears in. Yet I want each invocation/class to be able to customize the spying behavior somewhat. That's where implicits come (again) to the rescue. But the spy-function (the one that actually does something with a String formed from the expression's value) needs to be found. The natural type of the spy-function is String=>Unit but it turns out that the implicits mechanism couldn't deal with something so ordinary so I changed it to String=>Spy where the Spy class is essentially just a wrapper that has no real significance. Then, the implicit value for the spy-function (if any) could be found and used.

One example of a need to customize the implicit value is to forget about logging and simply write to the console. I tried to make this as easy as possible. See the specification in the repo (linked at the bottom) for an example of this.

Here, using the default slf4j logging mechanism, is the map2 function with logging. Note that, when you use the default mechanism, you must provide an implicit value of a logger in scope. A convenience method has been provided for this as shown below.

  implicit val logger = Spy.getLogger(getClass)
  def map2[T, U](to1: Option[T], to2: => Option[T])(f: (T, T) => U): Option[U] = Spy.spy(s"map2($to1,$to2)",for {t1 <- to1; t2 <- to2} yield f(t1, t2))

We test it using the following specification for map2 (unchanged):

  "map2(Option)" should "succeed" in {
    val one = Some(1)
    val two = Some(2)
    def sum(x: Int, y: Int) = x + y
    map2(one, two)(sum) should matchPattern { case Some(3) => }
    map2(one, None)(sum) should matchPattern { case None => }
  }

And the resulting entries in the log file are:

2016-12-06 22:35:04,123 DEBUG com.phasmid.laScala.fp.FP$  - spy: map2(Some(1),Some(2)): Some(3)
2016-12-06 22:35:04,128 DEBUG com.phasmid.laScala.fp.FP$  - spy: map2(Some(1),None): None

All we had to do, for the default logging behavior, was to ensure there was an implicit value of logger in scope and wrap the expression inside an invocation of Spy.spy. Everything is still purely functional. You can even leave the spy invocation in place if you really want to and either explicitly switch it off by adding false as a third parameter or by turning off debugging in the logger.

If you're interested in using this you can find the Spy class, together with its SpySpec, in my LaScala project.

A generic tail-recursive method for tree traversal

In an earlier article, Transforming iteration into tail recursion, I talked about the general functional-programming technique of replacing iteration by recursion. That method is quite simple and is fairly obvious for linear collections.

But this week, I was working on the general problem of tree traversal. In a binary tree, for instance, as you recurse through the tree, you need to keep track of the results of two recursive tree traversals at each node. They can't both be tail-recursive! But, fortunately, there is a well-known, more general technique available. The method basically consists, as usual, of creating an "inner" method with an accumulator (r for result) and a list of unprocessed elements (ts). The only extra complication is that when we visit a tree node, we need to add all the children (two in the case of a binary tree) to the unprocessed list at the same time as we process the node itself (if there is indeed anything at the node).

Because trees can take so many forms, I created a very general mechanism to implement this "inner" method, which I call traverse in the companion object of a type-class called Parent.

import scala.annotation.tailrec

/**
  * A trait which expresses parenthood.
  * This defines a type class.
  *
  * @tparam T the underlying type of the Parent
  */
trait Parent[T] {
  /**
    * Get the children of a T.
    *
    * @param t the parent whose children we want
    * @return the children as a Seq of T
    */
  def children(t: T): Seq[T]
}

object Parent {
  /**
    * Tail-recursive method to traverse a tree-like object (defined as something that implements Parent).
    *
    * @param f  the map function which takes a T and produces an R for just that parent object (not its children).
    * @param g  the reduce function which accumulates the result (the accumulator, i.e. r, is passed as its first parameter).
    * @param q  the quick return function which, if q(r) yields true, an immediate return of r is made.
    * @param z  the function which builds the T list (ts) from the existing T list and the given T (parent node).
    * @param ts a list of Ts to be worked on.
    * @param r  the current value of the result, i.e. the "accumulator".
    * @tparam T a type which extends Parent, and thus has children of type T -- this "context bound" is implemented via a compiler-generated implicit parameter of type Parent[T].
    * @tparam R the result type.
    * @return a value of R.
    */
  @tailrec
  final def traverse[T: Parent, R](f: T => R, g: (R, R) => R, q: R => Boolean, z: (List[T], T) => List[T])(ts: List[T], r: R): R =
    if (q(r))
      r
    else
      ts match {
        case Nil => r
        case h :: t => traverse(f, g, q, z)(z(t, h), g(r, f(h)))
      }
}

In order to use this for something like size, we need a tree structure made up of nodes, of type T, which implement Parent[T]. In the following code, Node[A] is a node in a tree.

/**
  * This trait expresses the notion of a Node from a tree.
  *
  */
sealed trait Node[+A] {
  /**
    * @return If this node contains an actual value x, return Some(x) else None
    */
  def get: Option[A]

  /**
    * Get the children of this Node, if any.
    * @return the children as a Seq of Nodes
    */
  def children: Seq[Node[A]]

  /**
    * @return the size of the subtree
    */
  def size: Int

  /**
    * @return true if node contains value or children; otherwise false
    */
  def isEmpty: Boolean

  /**
    * @param b the value to be searched
    * @tparam B the type of b
    * @return true if value b is found in the sub-tree defined by this
    */
  def includes[B >: A](b: B): Boolean
}

But, how can we make Node[A] extend Parent[Node[A]]? How about using the magic of type classes (see my earlier blog, Type classes, on the subject)? But a simple type class uses an implicit object which obviously can't itself have a parametric type.

The solution lies in an implicit conversion method. Normally, such methods convert from one type (T1) to another (T2) when the context calls for a T2 and only a T1 is available. However, in this case, T2 is Node[A] but there is no T1! But, the implicit mechanism works also for Unit=>T2. That's really cool!

So, here's the code that takes care of the implicit "evidence" for a node to be a parent:

object Node {
  /**
    * Implicit definition of a Parent for the (type-parameterized, i.e. generic) type Node[A]
    *
    * @tparam A the underlying node type
    * @return a Parent of type Node[A]
    */
  implicit def nodeParent[A]: Parent[Node[A]] = new Parent[Node[A]] {
    /**
      * @return the children as a Seq of Nodes
      */
    def children(t: Node[A]): Seq[Node[A]] = t.children
  }
}

And, finally, here is the code for the implementation of size and includes in a trait TreeLike[+A] which extends Node[A]:

  def size: Int = Parent.traverse[Node[A], Int]({
    case Empty => 0
    case _ => 1
  }, _+_, { x => false}, _ ++ _.children)(List(this), 0)

  def includes[B >: A](b: B): Boolean = Parent.traverse[Node[B], Boolean]({ n => n.get match {
    case Some(`b`) => true
    case _ => false
  }}, _|_, {x => x}, _ ++ _.children)(List(this), false)

Type classes come to the rescue again and we are able to define a wholly generic tail-recursive method for tree traversal.

Installation of PredictionIO on MacOS

I'm kicking the tires of PredictionIO, an open-source machine learning server. I'm hoping that this might be a great platform for my BigData/Scala class. It looks really promising.

However, kicking the tires is a lot easier at the car dealership. In order to get something like this running, there are many hurdles to jump. It would no doubt be easier if I was running on Windows, or if I wanted to install to virtual box or via Docker or something like that. The manual installation process for Mac took me several hours to crack as the error messages do not make it clear what the problem is. It's not unusual for an early-stage (incubating) project to have documentation/installation issues. On the whole, they've done an outstanding job. But, in case anyone else out there is having similar problems, here are my fixes to the installation docs. For the Mac (or linux), the installation procedure I followed is at Installing Apache PredictionIO (incubating) on Linux / Mac OS X.

PredictionIO is based on Spark, as well as some combination of storage solutions. You can use Postgres, MySQL or a combination of HBase and ElasticSearch. I chose the latter as it appears to be the recommended solution.

After following the instructions, including editing the conf/pio-env.sh
file to specify my own existing Spark installation, by editing SPARK_HOME as follows:

  SPARK_HOME=Applications/spark-1.6.1

I started the PIO suite using $PIO_HOME/bin/pio-start-all, which apparently went OK.

However, the first problem I ran into resulted in the following cryptic error message (when checking the status via pio status):

/Applications/spark-1.6.1 is probably an Apache Spark development tree. Please make sure you are using at least 1.3.0.

It turns out that you need a version of Spark pre-built with Hadoop (I had built this one from source). Before realizing that was the problem, I simply downloaded their recommended version into the vendors directory (rather than trying to put it in my own area) and edited the pio-env.sh file:

  SPARK_HOME=$PIO_HOME/vendors/spark-1.5.1-bin-hadoop2.6

Now, I tried it again. Again, getting the status showed a problem:

[ERROR] [Console$] Unable to connect to all storage backends successfully. The following shows the error message from the storage backend.

[ERROR] [Console$] Connection to localhost:5432 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections. (org.postgresql.util.PSQLException)

It isn't enough to set up the HBase and ElasticSearch configuration as instructed by the installation guide. You also have comment out the Postgres and MySQL variables and set up the PIO_STORAGE_REPOSITORIES_xxx variables as follows:

PIO_STORAGE_REPOSITORIES_METADATA_NAME=predictionio_metadata

PIO_STORAGE_REPOSITORIES_METADATA_SOURCE=ELASTICSEARCH

PIO_STORAGE_REPOSITORIES_EVENTDATA_NAME=predictionio_eventdata

PIO_STORAGE_REPOSITORIES_EVENTDATA_SOURCE=HBASE

PIO_STORAGE_REPOSITORIES_MODELDATA_NAME=pio_

PIO_STORAGE_REPOSITORIES_MODELDATA_SOURCE=LOCALFS

Then don't forget to configure local storage...

PIO_STORAGE_SOURCES_LOCALFS_TYPE=localfs

PIO_STORAGE_SOURCES_LOCALFS_PATH=$PIO_FS_BASEDIR/models

You also need to configure the PIO_STORAGE_SOURCES_xxx variables but the installation instructions already tell you to do that (I think).

Restarting and checking the status generated this error:

[ERROR] [RecoverableZooKeeper] ZooKeeper exists failed after 1 attempts

This took me a while to figure out. I actually installed ZooKeeper, although it turns out that I didn't need to -- and in fact it's incorrect to have ZooKeeper running if you're simply running HBase locally. The problem turned out to be that I hadn't properly configured HBase in the vendors/hbase-1.0.0/conf/hbase-site.xml configuration file. Look carefully at the values that you will find here. They do not match your setup!

Once I had fixed that problem, and rechecked the status, I saw the following wonderful message:

[INFO] [Console$] Your system is all ready to go.

You can also check that all is well by running jps. You should see something like the following:

41109 Console

41014 Elasticsearch

41079 HMaster

41127 Jps

29769 

If you don't see HMaster included, then HBase is not running correctly.

Now, it's time to clone a template. I'm interested in recommender systems so I navigated to the following set of instructions: which state in section 2. Create a new Engine from an Engine Template that you should run the following command:

pio template get PredictionIO/template-scala-parallel-recommendation MyRecommendation

However, that is (currently at least) incorrect. Instead, use the following:

pio template get PredictionIO/template-scala-parallel-universal-recommendation MyUniversalRecommendation

Once I figured out the correct mumbo-jumbo to clone the template, I was able to create my own recommendation engine. Now, I shall (hopefully) enjoy actually doing some implementation!

Stay tuned in case I have more tips.

Update 9/30/2016: there's one more thing you have to do, at least for the universal recommendation engine, and which is not spelled out in the installation instructions. Before you can import the sample data, you must install the Python predictionio package. Do this by entering the following:

pip install prediction

If that doesn't work for you, then consult this github project: https://github.com/apache/incubator-predictionio-sdk-python

One more problem which arose during the training phase:

[INFO] [Engine$] Data sanity check is on.

Exception in thread "main" java.lang.NumberFormatException: For input string: "558:feed::1"

at java.lang.NumberFormatException.forInputString(NumberFormatException.java:65)

The solution this time is something that I never would have imagined in a million years, let alone solved! Here's the link to the issue. But the bottom line is, if you're running on a Mac, change the TCP/IP wifi network settings so that Configure IPv6 is manual, not automatic.

And another problem after deploying and trying to make a query:

[ERROR] [OneForOneStrategy] Cannot support TLS_RSA_WITH_AES_256_CBC_SHA with currently installed providers

Here's a discussion of the issue.The solution is to upgrade your Java encryption policy to unlimited. Follow the instructions in the discussion. Ensure that you put the jars into the directory lib/security which lives under the JRE home, typically that would be at $JAVA_HOME/jre.

It turned out, however, that this particular solution was necessary but not sufficient. In the same discussion there's a mention of the SSL problem that I ran into next:

[ERROR] [HttpServerConnection] Aborting encrypted connection to /0:0:0:0:0:0:0:1:52194 due to [SSLException:Unrecognized SSL message, plaintext connection?]

You need to edit the shell script  examples/single-query-handmade.sh which sends queries to the PredictionIO server as follows (add "k" to the curl options and "s" to the http protocol value):

curl -kH "Content-Type: application/json" -d '

{

    "user": "u1"

}' https://localhost:8000/queries.json

Finally, after all this work, I was able to see the blessedly wonderful response from my recommendation engine:

Recommendations for user: u1

{"itemScores":[{"item":"Galaxy","score":0.8880454897880554},{"item":"Nexus","score":0.24007925391197205},{"item":"Surface","score":0.043848853558301926}]}