Errata

Greetings, I'm going through "Learning Spark" (3rd release, 1st edition).

Question on the top of top of page 29 of this Python example (example 3-15):

Print "Input had " + badLinesRDD.count() + " concerning lines"

I'm pretty sure this is wrong, because if you try that you get this error...


Traceback (most recent call last):

File "<stdin>", line 1, in <module>

TypeError: cannot concatenate 'str' and 'int' objects

... Python won't auto-convert ints to Strings like Java or Scala.


Context just in case I did something wrong in setting things up:

Douglass-MacBook-Pro-2:spark-1.4.1-bin-hadoop2.6 dmeil$ ./bin/pyspark

Python 2.7.10 |Anaconda 2.1.0 (x86_64)| (default, May 28 2015, 17:04:42)

[GCC 4.2.1 (Apple Inc. build 5577)] on darwin

Type "help", "copyright", "credits" or "license" for more information.

Anaconda is brought to you by Continuum Analytics.

Please check out: http://continuum.io/thanks and https://binstar.org

Welcome to

____ __

/ __/__ ___ _____/ /__

_\ \/ _ \/ _ `/ __/ '_/

/__ / .__/\_,_/_/ /_/\_\ version 1.4.1

/_/


Using Python version 2.7.10 (default, May 28 2015 17:04:42)

SparkContext available as sc, HiveContext available as sqlContext.

Note from the Author or Editor:
Thanks for catching this, I've added an explicit str :)

The R library Imap function gdist takes as the first four arguments longitude, latitude, longitude, and latitude. The calling code writes values to stdout in the wrong order (latitude, longitude, ...). This order is not corrected in the R code that passes them to gdist.

Examples 6-17 and 6-18 have the same bug.

Note from the Author or Editor:
That's correct, I've fixed this in 6-17/18 in atlas.

Default 'spark.sql.parquet.compression.codec' property is gzip.

https://spark.apache.org/docs/latest/sql-programming-guide.html#configuration

https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/SQLConf.scala#L251

val PARQUET_COMPRESSION = enumConf("spark.sql.parquet.compression.codec",
valueConverter = v => v.toLowerCase,
validValues = Set("uncompressed", "snappy", "gzip", "lzo"),
defaultValue = Some("gzip"),

Note from the Author or Editor:
This is correct (although in the current version of Spark it's now snappy). I've updated the table to mention this.

READ:
JavaSparkContext sc = JavaSparkContext(conf);

SHOULD READ:
JavaSparkContext sc = new JavaSparkContext(conf);

Note from the Author or Editor:
Thanks, I've fixed this in atlas.

Example 8-1. Creating an application using a SparkConf in Python

1) You should remove the 'new' in conf = new SparkConf()
2) You should change the argument in SparkContext()
https://spark.apache.org/docs/latest/programming-guide.html#initializing-spark
https://issues.apache.org/jira/browse/SPARK-2003

Correct code would be:

conf = SparkConf()
conf.setAppName("My Spark App")
conf.set("spark.master", "local[4]")
conf.set("spark.ui.port", "36000")

sc = SparkContext(conf=conf)

OR

conf = SparkConf() \
.setAppName("My Spark App") \
.set("spark.master", "local[4]") \
.set("spark.ui.port", "36000")

sc = SparkContext(conf=conf)

Note from the Author or Editor:
Thanks for catching this, I've fixed it in our dev version.

In last two lines, "Durations" is misprinted as "Dirations".

Note from the Author or Editor:
This is correct, I've fixed this in atlas.

Page 64, Section "Determining an RDD’s Partitioner", second line says, "or partitioner() method in Java".
There is no method "partitioner()" available on "org.apache.spark.api.java.JavaPairRDD." (spark version 1.3.1)
Is this a typo for the method "partitions()"?

Note from the Author or Editor:
Seems that the partitioner() function doesn't exist. I'll drop it. (partitions() doesn't quite return the partitioner rather a list of the partitions).

collectAsMap() doesn't return multi Map. so Result should be
Map{(1,2), (3, 6)}

https://github.com/apache/spark/blob/b0d884f044fea1c954da77073f3556cd9ab1e922/core/src/main/scala/org/apache/spark/rdd/PairRDDFunctions.scala#L659

/**
* Return the key-value pairs in this RDD to the master as a Map.
*
* Warning: this doesn't return a multimap (so if you have multiple values to the same key, only
* one value per key is preserved in the map returned)

Note from the Author or Editor:
Thanks for catching that, I've updated the example.

No semicolon ends.

READ:
System.out.println("Input had " + badLinesRDD.count() + " concerning lines") System.out.println("Here are 10 examples:")

SHOULD READ:
System.out.println("Input had " + badLinesRDD.count() + " concerning lines"); System.out.println("Here are 10 examples:");

Note from the Author or Editor:
Thanks for catching this, I've added in the missing semicolons to example 3-17.

Python example : Example 3-15

below line return TypeError.

READ:
print "Input had " + badLinesRDD.count() + " concerning lines"
---------------------------------------------------------------------------
TypeError Traceback (most recent call last)
<ipython-input-61-13cf6c82420f> in <module>()
----> 1 print "Input had " + badLinesRDD.count() + " concerning lines"

"TypeError: cannot concatenate 'str' and 'int' objects"

SHOULD READ:

print "Input had %i" % badLinesRDD.count() + " concerning lines"

or

print "Input had " + str(badLinesRDD.count()) + " concerning lines"

Note from the Author or Editor:
Thanks, this has been fixed.

Example 10-5 @override is missing on the call method.

Note from the Author or Editor:
Added missing @override

spark.[X}.port explanation is missing the ui value (spark.ui.port)

Note from the Author or Editor:
Yup we can configure the UI this way too, updated the table.

std.stdev in example 6-19 should be stats.stdev

Note from the Author or Editor:
Fixed in dev :)

String interpolation in example 6-17 needs to be in brackets as it uses a property of the y object.

Note from the Author or Editor:
Good catch, fixed in the dev build.

"If there are only a few input files, and you need to use the wholeFile() method,"

should be

"If there are only a few input files, and you need to use the wholeTextFile() method,"

Note from the Author or Editor:
Thanks, fixed to wholeTextFiles.

import Java.io.StringReader should use a lowercase j

This happens in a number of locations actually.

Note from the Author or Editor:
Thanks for catching this. I've applied a global fix to the dev copy.

The case under //Run 10 iterations shadows the links variable. This might be confusing for new developers

Note from the Author or Editor:
Thanks, thats a good point since that behaviour is maybe confusing for people coming from other languages. I've clarified it by using a different local variable in the dev verison.

In Figure 10-7, An arrow is missing on the inverse graph from {4,2} to 20

Note from the Author or Editor:
On the right side of the graph we need to add an arrow from the {4,2} box to 20 and remove its line to box 22.

Example 8-11 coalesces to 5, but the number of partitions is listed as 4.

There are couple of naming errors in Scala version of example for newer (as of Spark 1.3) Spark Streaming and Apache Kafka createDirectStream() method (Example 10-34: Apache Kafka directly reading Panda's topic in Scala):

- brokers val should be renamed to kafkaParams (brokers is referenced inside Map.apply() method);
- topics val should be renamed to topicsSet (topicSet is referenced inside createDirectStream() method);

Note from the Author or Editor:
Thanks for catching this, I've fixed it it in the latest dev version.

In example 3-40, "result.persist(StorageLevel.DISK_ONLY)" will not work, as it is not imported in the example.

Adding "import org.apache.spark.storage.StorageLevel" will fix this.

Note from the Author or Editor:
Thanks for pointing this out, we mention the package that the StorageLevels come from in the persistence table but I've added an import in the example code for clarity.

I have been earlier asked by the author to return the book because I reported an issue.

I am therefore not writing this with the intention of it being corrected, but just to let others know that how the code can be run correctly.

The following code does not run:
-----------------------------------------------------------------------------------------------
class SearchFunctions(val query: String) { def isMatch(s: String): Boolean = {

s.contains(query)
}
def getMatchesFunctionReference(rdd: RDD[String]): RDD[String] = {
// Problem: "isMatch" means "this.isMatch", so we pass all of "this”

rdd.map(isMatch)

}
def getMatchesFieldReference(rdd: RDD[String]): RDD[String] = {

// Problem: "query" means "this.query", so we pass all of "this"
rdd.map(x => x.split(query)) }

def getMatchesNoReference(rdd: RDD[String]): RDD[String] = {
// Safe: extract just the field we need into a local variable

val query_ = this.query
rdd.map(x => x.split(query_))

} }

-----------------------------------------------------------------------------------------------


And it can be modified or updated to the following so that it can run:
-----------------------------------------------------------------------------------------------
// As the RDD class is not automatically imported therefore we have to import it explicitly
import org.apache.spark.rdd.RDD

class SearchFunctions(val query: String) {
def isMatch(s : String ): Boolean = {
s.contains(query)
}

def getMatchesFunctionReference(rdd: RDD[String]): RDD[Boolean] = {
// Problem: "isMatch" means "this.isMatch", so we pass all of "this"
rdd.map(isMatch)
}

def getMatchesFieldReference(rdd: RDD[String]): RDD[Array[String]] = {
// Problem: "query" means "this.query", so we pass all of "this"
rdd.map(x => x.split(query))

}

def getMatchesNoReference(rdd: RDD[String]): RDD[Array[String]] = {
// Safe: extract just the field we need into a local variable
val query_ = this.query
rdd.map(x => x.split(query_))
}
}
-----------------------------------------------------------------------------------------------


Regards,
Gourav

Note from the Author or Editor:
We should include the import org.apache.spark.rdd.RDD in the standard imports.

Example 5-20. Loading a SequenceFile in Python should drop the "val" on "val data = ..." Works otherwise.

Note from the Author or Editor:
Thanks for catching this, I went ahead and fixed this in atlass.

Shouldn't "rdd.flatMap(...)" be "input.flatMap(...)"

Note from the Author or Editor:
Fixed in atlass

In example 4-24, "val partitioned = pairs.partitionBy(new spark.HashPartitioner(2))" will not work, as it is not imported in the example. Either an import or a change into "new org.apache.spark.HashPartitioner(2)" would work.

Note from the Author or Editor:
Thanks for pointing this out, I'll update our example to include the import. Fixed in cd090206381a9bbf0466468bf7128a808085522f.

It says "(...) run bin/stop-all.sh (...)"
It should be "(...) run sbin/stop-all.sh (...)"

Note from the Author or Editor:
Thanks for catching that, I've updated it in adbdb12def7218b0f54cb67f96cb688775e05ec5

(p. 109 PDF document; page 91 of book). Minor -- with the import of the HiveContext class, there's no need to fully qualify the class name when invoking the HiveContext constructor.

Note from the Author or Editor:
Thanks for catching this, I've simplified the code as suggested in b9d7e376aae27e2f8d4de6d431691a62852d92ba.

(p. 91 of the PDF doc; p. 73 of the book). This is a total nitpick, but the file url is
file://home/holden/salesFiles
and instead should be
file:///home/holden/salesFiles

Note from the Author or Editor:
Thanks, fixed :)

Apparent cut-and-paste mistake: the "Custom parallelism" example is the same as the default one, in that no parallelism Int was specified in the example call.

Note from the Author or Editor:
Fixed, thanks :)

In Example 3-21, return types for the getMatches* methods are incorrect. getMatchesFunctionReference() should return RDD[Boolean], and getMatchesFieldReference() and getMatchesNoReference() should either return RDD[Array[String]] or change the implementation to use flatMap instead of map.

Note from the Author or Editor:
I've updated the return types, thanks for catching this.

"Example 2-13. Maven build file" is invalid because there is an extra </plugin> closing tag.

Bad: </configuration> </plugin> </plugin>
Better: </configuration> </plugin>

Note from the Author or Editor:
Removed extra plugin tag (done by author).

Python aggregate example
sumCount = nums.aggregate((0, 0),
(lambda acc, value: (acc[0] + value, acc[1] + 1)), >>add missing paren<<
(lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])))

Last line remove extra parenthesis

Note from the Author or Editor:
Fixed by author in 54759cf2cf0e41b81bdd56eaa5adb308ac911845

"We have looked at the fold, combine, and reduce actions on basic RDDs". There is no RDD.combine(), did you mean aggregate()?

Note from the Author or Editor:
Replace combine with aggregate (fixed in f7df06b0c1d730a3a20f173dea8d4ce5c137aa0d).

Example 4-12 (Python) is not equivalent to the others: the sum of numbers must be divided by the count to yield the average. Having the Python example implement the same behavior as the Scala and Java examples will aid the reader. My version of the example is:

nums = sc.parallelize([(1,2),(1,4),(3,6),(4,6),(4,8),(4,13)])
sumCount = nums.combineByKey((lambda x : (x , 1)),
(lambda x, y : (x[0] + y, x[1] + 1)),
(lambda x ,y : (x [0] + y[0], x[1] + y[1])))
print sumCount.map(lambda (k,v): (k, v[0]/float(v[1]))).collect()

Note from the Author or Editor:
no action, fixed in book already.

<artifactId>learning-spark-mini-example/artifactId> is missing closing <

Note from the Author or Editor:
Fixed in b99b12fcd3022c298d30f3fcd2b1d88fd7eab57c

Minor issue; there should be a

import Java.io.StringReader

statement in your CSV loading examples in Scala (and presumably Java)

Note from the Author or Editor:
I fixed in holden@hmbp2:~/repos/1230000000573$ git log
commit a9f9f34a3b8513885325f47c1101e657cb5faa89

Maven command line executable is called 'mvn'. Please replace "maven clean && maven compile && maven package" with "mvn clean && mvn compile && mvn package".

Also the maven build script (Example 2-13) doesn't compile scala code (i.e. c.o.l.mini.scala), please replace

"$SPARK_HOME/bin/spark-submit --class com.oreilly.learningsparkexamples.mini.scala.WordCount \
./target/learning-spark-mini-example-0.0.1.jar ./README.md ./wordcounts"

with

"$SPARK_HOME/bin/spark-submit --class com.oreilly.learningsparkexamples.mini.java.WordCount \
./target/learning-spark-mini-example-0.0.1.jar ./README.md ./wordcounts"

Note from the Author or Editor:
Fixed

println(result.collect())

should be

result.collect().foreach{x=>println(x)}

Note from the Author or Editor:
Fixed in the latest build.

FilteredRDD
{1,4,9,16}

should be

FilteredRDD
{2,3,4}

Note from the Author or Editor:
Thanks for pointing this out, I've gone ahead and fixed this and it should be in our next build.

It is mentioned that liftweb-json is used for JSON-parsing, however Play JSON is used for parsing and then liftweb-json for JSON output. This is a bit confusing.

Note from the Author or Editor:
I've fixed this in the latest push.