Pyspark Recipes

Counting Lines

This program will load the Google Unigrams dataset and count the total number of lines:

from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName("Unigrammer3").setMaster("local[16]").set("spark.executor.memory", "4g")
sc = SparkContext(conf=conf)

lines = sc.textFile("hdfs://master:9000/dat/GG/eng/01")
count = lines.count()
print "total lines %i" % (count)

Create an empty DataFrame with a given schema:

%pyspark

from pyspark.sql.types import *
field = [
    StructField("userid", IntegerType(), True), 
    StructField("posted_time", DateType(), True),
    StructField("category", StringType(), True),
    StructField("parent", StringType(), True),
    StructField("entity", StringType(), True),
    StructField("tweetid", IntegerType(), True),
    StructField("content", StringType(), True),
    StructField("lang", StringType(), True),
    StructField("search_object", StringType(), True),
    StructField("sony", BooleanType(), True),
    StructField("spam", BooleanType(), True),
    StructField("spam_match", StringType(), True),
    StructField("thandle", StringType(), True),
    StructField("turner", BooleanType(), True),
    
    ]
schema = StructType(field)

df = sqlContext.createDataFrame(sc.emptyRDD(), schema)

Merge multiple data frames:

%pyspark

from functools import reduce  # For Python 3.x
from pyspark.sql import DataFrame

def unionAll(*dfs):
    return reduce(DataFrame.unionAll, dfs)
    
df = unionAll(
    df_fb[df_fb.entity == "despicable me 1"],
    df_fb[df_fb.entity == "elysium"],
    df_fb[df_fb.entity == "evil dead"])
df.limit(2).show()

Deploy a Scala job to a Spark Cluster using SBT

Environment

1. sbt 0.13.8
2. Scala 2.11.6
3. Spark 1.2.1
4. Debian Linux (Ubuntu 14.10)

Scala Program

This simple program written in Scala will analyze a local file on my system, count the number of times that lines containing a and b occur, and print the total to the console.

/*** SimpleApp.scala ***/
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

object SimpleApp {
  def main(args: Array[String]) {
    val logFile = "/home/craig/spark/README.md" // Should be some file on your system
    val sc = new SparkContext("local", "Simple App", "$SPARK_HOME", List("target/scala-2.11/simple-project_2.11-1.0.jar"))
    val logData = sc.textFile(logFile, 2).cache()
    val numAs = logData.filter(line => line.contains("a")).count()
    val numBs = logData.filter(line => line.contains("b")).count()
    println("Lines with a: %s, Lines with b: %s".format(numAs, numBs))
  }
}

Directory Structure

I've created the following directory structure in my home directory:

mkdir -p ~/apps/simple/src/main/scala

The scala program above will be saved under the "src/main/scala" directory.

Build Script

Create a build script called simple.sbt under the "~/apps/simple/" directory:

name := "Simple Project"

version := "1.0"

scalaVersion := "2.11.6"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.2.1"
libraryDependencies += "org.apache.hadoop" % "hadoop-client" % "2.6.0"

resolvers += "Akka Repository" at "http://repo.akka.io/releases/"

The lines highlighted in red should be replaced with your specific version numbers.

Building the Program

Navigate to ~/apps/simple and type "sbt" on the terminal session:

craig@spark:~/apps/simple$ sbt
[info] Set current project to Simple Project (in build file:/home/craig/apps/simple/)
> 

Type "package" at the prompt:

> package
[info] Updating {file:/home/craig/apps/simple/}simple...
[info] Resolving jline#jline;2.12.1 ...
[info] Done updating.
[info] Compiling 1 Scala source to /home/craig/apps/simple/target/scala-2.11/classes...
[info] Packaging /home/craig/apps/simple/target/scala-2.11/simple-project_2.11-1.0.jar ...
[info] Done packaging.
[success] Total time: 18 s, completed Mar 24, 2015 11:15:18 AM

My directory structure now looks like this:

craig@spark:~/apps/simple$ tree
.
├── simple.sbt
├── src
│   └── main
│       └── scala
│           └── SimpleApp.scala
└── target
    ├── resolution-cache
    │   ├── reports
    │   │   ├── ivy-report.css
    │   │   ├── ivy-report.xsl
    │   │   ├── simple-project-simple-project_2.11-compile-internal.xml
    │   │   ├── simple-project-simple-project_2.11-compile.xml
    │   │   ├── simple-project-simple-project_2.11-docs.xml
    │   │   ├── simple-project-simple-project_2.11-optional.xml
    │   │   ├── simple-project-simple-project_2.11-plugin.xml
    │   │   ├── simple-project-simple-project_2.11-pom.xml
    │   │   ├── simple-project-simple-project_2.11-provided.xml
    │   │   ├── simple-project-simple-project_2.11-runtime-internal.xml
    │   │   ├── simple-project-simple-project_2.11-runtime.xml
    │   │   ├── simple-project-simple-project_2.11-scala-tool.xml
    │   │   ├── simple-project-simple-project_2.11-sources.xml
    │   │   ├── simple-project-simple-project_2.11-test-internal.xml
    │   │   └── simple-project-simple-project_2.11-test.xml
    │   └── simple-project
    │       └── simple-project_2.11
    │           └── 1.0
    │               ├── resolved.xml.properties
    │               └── resolved.xml.xml
    ├── scala-2.11
    │   ├── classes
    │   │   ├── SimpleApp$$anonfun$1.class
    │   │   ├── SimpleApp$$anonfun$2.class
    │   │   ├── SimpleApp.class
    │   │   └── SimpleApp$.class
    │   └── simple-project_2.11-1.0.jar
    └── streams
        ├── compile
        │   ├── compile
        │   │   └── $global
        │   │       └── streams
        │   │           └── out
        │   ├── compileIncremental
        │   │   └── $global
        │   │       └── streams
        │   │           ├── export
        │   │           └── out
        │   ├── copyResources
        │   │   └── $global
        │   │       └── streams
        │   │           ├── copy-resources
        │   │           └── out
        │   ├── dependencyClasspath
        │   │   └── $global
        │   │       └── streams
        │   │           └── export
        │   ├── externalDependencyClasspath
        │   │   └── $global
        │   │       └── streams
        │   │           └── export
        │   ├── $global
        │   │   └── $global
        │   │       └── discoveredMainClasses
        │   │           └── data
        │   ├── incCompileSetup
        │   │   └── $global
        │   │       └── streams
        │   │           └── inc_compile_2.11
        │   ├── internalDependencyClasspath
        │   │   └── $global
        │   │       └── streams
        │   │           └── export
        │   ├── mainClass
        │   │   └── $global
        │   │       └── streams
        │   │           └── out
        │   ├── managedClasspath
        │   │   └── $global
        │   │       └── streams
        │   │           └── export
        │   ├── packageBin
        │   │   └── $global
        │   │       └── streams
        │   │           ├── inputs
        │   │           ├── out
        │   │           └── output
        │   ├── unmanagedClasspath
        │   │   └── $global
        │   │       └── streams
        │   │           └── export
        │   └── unmanagedJars
        │       └── $global
        │           └── streams
        │               └── export
        └── $global
            ├── dependencyPositions
            │   └── $global
            │       └── streams
            │           └── update_cache_2.11
            │               ├── input_dsp
            │               └── output_dsp
            ├── $global
            │   └── $global
            │       └── streams
            │           └── out
            ├── ivyConfiguration
            │   └── $global
            │       └── streams
            │           └── out
            ├── ivySbt
            │   └── $global
            │       └── streams
            │           └── out
            ├── projectDescriptors
            │   └── $global
            │       └── streams
            │           └── out
            └── update
                └── $global
                    └── streams
                        ├── out
                        └── update_cache_2.11
                            ├── inputs
                            └── output

73 directories, 50 files

Running the Program

Inside your working directory (~/apps/simple), execute this command:

$SPARK_HOME/bin/./spark-submit --class "SimpleApp" --master local[8] target/scala-2.11/simple-project_2.11-1.0.jar

Successful output on my workstation looks like this:

Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
15/03/24 11:20:37 INFO SecurityManager: Changing view acls to: craig
15/03/24 11:20:37 INFO SecurityManager: Changing modify acls to: craig
15/03/24 11:20:37 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(craig); users with modify permissions: Set(craig)
15/03/24 11:20:38 INFO Slf4jLogger: Slf4jLogger started
15/03/24 11:20:38 INFO Remoting: Starting remoting
15/03/24 11:20:38 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriver@10.0.4.15:49222]
15/03/24 11:20:38 INFO Utils: Successfully started service 'sparkDriver' on port 49222.
15/03/24 11:20:38 INFO SparkEnv: Registering MapOutputTracker
15/03/24 11:20:38 INFO SparkEnv: Registering BlockManagerMaster
15/03/24 11:20:38 INFO DiskBlockManager: Created local directory at /tmp/spark-91e5f424-b1e6-4d51-a010-a4e0ac788725/spark-2d07771d-f3ad-4c70-bb5b-1329be484b4f
15/03/24 11:20:38 INFO MemoryStore: MemoryStore started with capacity 265.1 MB
15/03/24 11:20:38 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/03/24 11:20:38 INFO HttpFileServer: HTTP File server directory is /tmp/spark-cf38f85d-efd0-4941-b4f6-d245b9d8380c/spark-f903d7b0-c67c-48f0-9eb0-4a3279903d2a
15/03/24 11:20:38 INFO HttpServer: Starting HTTP Server
15/03/24 11:20:39 INFO Utils: Successfully started service 'HTTP file server' on port 50465.
15/03/24 11:20:39 INFO Utils: Successfully started service 'SparkUI' on port 4040.
15/03/24 11:20:39 INFO SparkUI: Started SparkUI at http://10.0.4.15:4040
15/03/24 11:20:39 INFO SparkContext: Added JAR target/scala-2.11/simple-project_2.11-1.0.jar at http://10.0.4.15:50465/jars/simple-project_2.11-1.0.jar with timestamp 1427221239211
15/03/24 11:20:39 INFO Executor: Starting executor ID <driver> on host localhost
15/03/24 11:20:39 INFO AkkaUtils: Connecting to HeartbeatReceiver: akka.tcp://sparkDriver@10.0.4.15:49222/user/HeartbeatReceiver
15/03/24 11:20:39 INFO NettyBlockTransferService: Server created on 54739
15/03/24 11:20:39 INFO BlockManagerMaster: Trying to register BlockManager
15/03/24 11:20:39 INFO BlockManagerMasterActor: Registering block manager localhost:54739 with 265.1 MB RAM, BlockManagerId(<driver>, localhost, 54739)
15/03/24 11:20:39 INFO BlockManagerMaster: Registered BlockManager
15/03/24 11:20:39 INFO MemoryStore: ensureFreeSpace(180608) called with curMem=0, maxMem=278019440
15/03/24 11:20:39 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 176.4 KB, free 265.0 MB)
15/03/24 11:20:40 INFO MemoryStore: ensureFreeSpace(25432) called with curMem=180608, maxMem=278019440
15/03/24 11:20:40 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 24.8 KB, free 264.9 MB)
15/03/24 11:20:40 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost:54739 (size: 24.8 KB, free: 265.1 MB)
15/03/24 11:20:40 INFO BlockManagerMaster: Updated info of block broadcast_0_piece0
15/03/24 11:20:40 INFO SparkContext: Created broadcast 0 from textFile at SimpleApp.scala:9
15/03/24 11:20:40 INFO FileInputFormat: Total input paths to process : 1
15/03/24 11:20:40 INFO SparkContext: Starting job: count at SimpleApp.scala:10
15/03/24 11:20:40 INFO DAGScheduler: Got job 0 (count at SimpleApp.scala:10) with 2 output partitions (allowLocal=false)
15/03/24 11:20:40 INFO DAGScheduler: Final stage: Stage 0(count at SimpleApp.scala:10)
15/03/24 11:20:40 INFO DAGScheduler: Parents of final stage: List()
15/03/24 11:20:40 INFO DAGScheduler: Missing parents: List()
15/03/24 11:20:40 INFO DAGScheduler: Submitting Stage 0 (FilteredRDD[2] at filter at SimpleApp.scala:10), which has no missing parents
15/03/24 11:20:40 INFO MemoryStore: ensureFreeSpace(2720) called with curMem=206040, maxMem=278019440
15/03/24 11:20:40 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 2.7 KB, free 264.9 MB)
15/03/24 11:20:40 INFO MemoryStore: ensureFreeSpace(1950) called with curMem=208760, maxMem=278019440
15/03/24 11:20:40 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 1950.0 B, free 264.9 MB)
15/03/24 11:20:40 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on localhost:54739 (size: 1950.0 B, free: 265.1 MB)
15/03/24 11:20:40 INFO BlockManagerMaster: Updated info of block broadcast_1_piece0
15/03/24 11:20:40 INFO SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:838
15/03/24 11:20:40 INFO DAGScheduler: Submitting 2 missing tasks from Stage 0 (FilteredRDD[2] at filter at SimpleApp.scala:10)
15/03/24 11:20:40 INFO TaskSchedulerImpl: Adding task set 0.0 with 2 tasks
15/03/24 11:20:40 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, PROCESS_LOCAL, 1361 bytes)
15/03/24 11:20:40 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
15/03/24 11:20:40 INFO Executor: Fetching http://10.0.4.15:50465/jars/simple-project_2.11-1.0.jar with timestamp 1427221239211
15/03/24 11:20:40 INFO Utils: Fetching http://10.0.4.15:50465/jars/simple-project_2.11-1.0.jar to /tmp/spark-4055924d-8bde-4216-9932-526612364a63/spark-ac26b090-87ab-4f55-b0b9-3c8571d43307/fetchFileTemp507742889710724975.tmp
15/03/24 11:20:40 INFO Executor: Adding file:/tmp/spark-4055924d-8bde-4216-9932-526612364a63/spark-ac26b090-87ab-4f55-b0b9-3c8571d43307/simple-project_2.11-1.0.jar to class loader
15/03/24 11:20:40 INFO CacheManager: Partition rdd_1_0 not found, computing it
15/03/24 11:20:40 INFO HadoopRDD: Input split: file:/home/craig/spark/README.md:0+1814
15/03/24 11:20:40 INFO deprecation: mapred.tip.id is deprecated. Instead, use mapreduce.task.id
15/03/24 11:20:40 INFO deprecation: mapred.task.id is deprecated. Instead, use mapreduce.task.attempt.id
15/03/24 11:20:40 INFO deprecation: mapred.task.is.map is deprecated. Instead, use mapreduce.task.ismap
15/03/24 11:20:40 INFO deprecation: mapred.task.partition is deprecated. Instead, use mapreduce.task.partition
15/03/24 11:20:40 INFO deprecation: mapred.job.id is deprecated. Instead, use mapreduce.job.id
15/03/24 11:20:40 INFO MemoryStore: ensureFreeSpace(6208) called with curMem=210710, maxMem=278019440
15/03/24 11:20:40 INFO MemoryStore: Block rdd_1_0 stored as values in memory (estimated size 6.1 KB, free 264.9 MB)
15/03/24 11:20:40 INFO BlockManagerInfo: Added rdd_1_0 in memory on localhost:54739 (size: 6.1 KB, free: 265.1 MB)
15/03/24 11:20:40 INFO BlockManagerMaster: Updated info of block rdd_1_0
15/03/24 11:20:40 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2326 bytes result sent to driver
15/03/24 11:20:40 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1, localhost, PROCESS_LOCAL, 1361 bytes)
15/03/24 11:20:40 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
15/03/24 11:20:40 INFO CacheManager: Partition rdd_1_1 not found, computing it
15/03/24 11:20:40 INFO HadoopRDD: Input split: file:/home/craig/spark/README.md:1814+1815
15/03/24 11:20:40 INFO MemoryStore: ensureFreeSpace(5400) called with curMem=216918, maxMem=278019440
15/03/24 11:20:40 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 260 ms on localhost (1/2)
15/03/24 11:20:40 INFO MemoryStore: Block rdd_1_1 stored as values in memory (estimated size 5.3 KB, free 264.9 MB)
15/03/24 11:20:40 INFO BlockManagerInfo: Added rdd_1_1 in memory on localhost:54739 (size: 5.3 KB, free: 265.1 MB)
15/03/24 11:20:40 INFO BlockManagerMaster: Updated info of block rdd_1_1
15/03/24 11:20:40 INFO Executor: Finished task 1.0 in stage 0.0 (TID 1). 2326 bytes result sent to driver
15/03/24 11:20:40 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 37 ms on localhost (2/2)
15/03/24 11:20:40 INFO DAGScheduler: Stage 0 (count at SimpleApp.scala:10) finished in 0.299 s
15/03/24 11:20:40 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
15/03/24 11:20:40 INFO DAGScheduler: Job 0 finished: count at SimpleApp.scala:10, took 0.448259 s
15/03/24 11:20:40 INFO SparkContext: Starting job: count at SimpleApp.scala:11
15/03/24 11:20:40 INFO DAGScheduler: Got job 1 (count at SimpleApp.scala:11) with 2 output partitions (allowLocal=false)
15/03/24 11:20:40 INFO DAGScheduler: Final stage: Stage 1(count at SimpleApp.scala:11)
15/03/24 11:20:40 INFO DAGScheduler: Parents of final stage: List()
15/03/24 11:20:40 INFO DAGScheduler: Missing parents: List()
15/03/24 11:20:40 INFO DAGScheduler: Submitting Stage 1 (FilteredRDD[3] at filter at SimpleApp.scala:11), which has no missing parents
15/03/24 11:20:40 INFO MemoryStore: ensureFreeSpace(2720) called with curMem=222318, maxMem=278019440
15/03/24 11:20:40 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 2.7 KB, free 264.9 MB)
15/03/24 11:20:40 INFO MemoryStore: ensureFreeSpace(1950) called with curMem=225038, maxMem=278019440
15/03/24 11:20:40 INFO BlockManager: Removing broadcast 1
15/03/24 11:20:40 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 1950.0 B, free 264.9 MB)
15/03/24 11:20:40 INFO BlockManager: Removing block broadcast_1
15/03/24 11:20:40 INFO MemoryStore: Block broadcast_1 of size 2720 dropped from memory (free 277795172)
15/03/24 11:20:40 INFO BlockManager: Removing block broadcast_1_piece0
15/03/24 11:20:40 INFO MemoryStore: Block broadcast_1_piece0 of size 1950 dropped from memory (free 277797122)
15/03/24 11:20:40 INFO BlockManagerInfo: Added broadcast_2_piece0 in memory on localhost:54739 (size: 1950.0 B, free: 265.1 MB)
15/03/24 11:20:40 INFO BlockManagerMaster: Updated info of block broadcast_2_piece0
15/03/24 11:20:40 INFO BlockManagerInfo: Removed broadcast_1_piece0 on localhost:54739 in memory (size: 1950.0 B, free: 265.1 MB)
15/03/24 11:20:40 INFO SparkContext: Created broadcast 2 from broadcast at DAGScheduler.scala:838
15/03/24 11:20:40 INFO BlockManagerMaster: Updated info of block broadcast_1_piece0
15/03/24 11:20:40 INFO DAGScheduler: Submitting 2 missing tasks from Stage 1 (FilteredRDD[3] at filter at SimpleApp.scala:11)
15/03/24 11:20:40 INFO TaskSchedulerImpl: Adding task set 1.0 with 2 tasks
15/03/24 11:20:40 INFO ContextCleaner: Cleaned broadcast 1
15/03/24 11:20:40 INFO TaskSetManager: Starting task 0.0 in stage 1.0 (TID 2, localhost, PROCESS_LOCAL, 1361 bytes)
15/03/24 11:20:40 INFO Executor: Running task 0.0 in stage 1.0 (TID 2)
15/03/24 11:20:40 INFO BlockManager: Found block rdd_1_0 locally
15/03/24 11:20:40 INFO Executor: Finished task 0.0 in stage 1.0 (TID 2). 1757 bytes result sent to driver
15/03/24 11:20:40 INFO TaskSetManager: Starting task 1.0 in stage 1.0 (TID 3, localhost, PROCESS_LOCAL, 1361 bytes)
15/03/24 11:20:40 INFO Executor: Running task 1.0 in stage 1.0 (TID 3)
15/03/24 11:20:40 INFO TaskSetManager: Finished task 0.0 in stage 1.0 (TID 2) in 25 ms on localhost (1/2)
15/03/24 11:20:40 INFO BlockManager: Found block rdd_1_1 locally
15/03/24 11:20:40 INFO Executor: Finished task 1.0 in stage 1.0 (TID 3). 1757 bytes result sent to driver
15/03/24 11:20:40 INFO TaskSetManager: Finished task 1.0 in stage 1.0 (TID 3) in 15 ms on localhost (2/2)
15/03/24 11:20:40 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks have all completed, from pool 
15/03/24 11:20:40 INFO DAGScheduler: Stage 1 (count at SimpleApp.scala:11) finished in 0.040 s
15/03/24 11:20:40 INFO DAGScheduler: Job 1 finished: count at SimpleApp.scala:11, took 0.067147 s
Lines with a: 60, Lines with b: 29

References

1. [Spark] Quick Start Guide
1. Contains the SimpleApp.scala program that was modified for this article

Spark Architecture and Design

Cluster Mode Overview

Spark applications run as independent sets of processes on a cluster, coordinated by the SparkCOntext object in your main program (aka Driver Program).

Fig 1: Cluster Mode Overview

What is RDD?

Write programs in terms of transformations on distributed datasets.

Resilient Distributed Datasets
1. Collections of objects spread across a cluster, stored in RAM or on Disk
2. Built through parallel transformations
3. Automatically rebuilt on failure Operations

Operations:
1. Transformations (e.g. map, filter, groupBy)
2. Actions (e.g. count, collect, save)

References

1. Cluster Design
1. [Spark] Cluster Mode Overview
2. RDD
1. The RDD API by Example
1. Zhen He's page at La Trobe University.  
2. Current with Spark 1.1.0
3. A helpful introduction to the RDD API.
2. [DataBricks, PDF] Spark Tutorial Summit 2013
1. Introductory level talk

Installing Spark on Ubuntu

Why Spark?

Spark comes with tools for interactive query analysis (Shark), large-scale graph processing and analysis (Bagel), and real-time analysis (Spark Streaming). Rather than having to mix and match a set of tools (e.g., Hive, Hadoop, Mahout, S4/Storm), you only have to learn one programming paradigm. For SQL enthusiasts, the added bonus is that Shark tends to run faster than Hive.

Environment

1. Apache Spark 1.2.1
2. Ubuntu 14.10
3. Git
4. JDK 1.8.0_31

Make sure you have Java and Git installed.  Double check that your $JAVA_HOME environment variable points to a valid jdk and not a jre.

1. Installing Spark

The following commands have been tested, and are operational, under Ubuntu 14.10.

mkdir ~/spark
cd ~/spark
wget http://d3kbcqa49mib13.cloudfront.net/spark-1.2.1.tgz
tar -zxvf spark-1.2.1.tgz

I don't recommend using a directory installation path which will require the use of sudo privileges to install spark.  I attempted this initially, and ran into an Access Control Exception when trying to connect Spark to my existing HDFS cluster.

2. Modifying the Path

Modify both the PATH and CLASSPATH to point to the new Scala installation.

Use the text editor of your choice to edit your environment file:

sudo gedit /etc/environment

The italicized text was added:

PATH="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/usr/lib/apache/spark/1.2.1/bin"
 ...   
export SPARK_HOME=/usr/lib/apache/spark/1.2.1

Once the environment file is saved, reload it:

source /etc/environment

3. Verifying the Installation

We're going to run a sample program that is bundled with Spark to verify the installation was performed correctly.

The program is called SparkPi and will output a value for Pi. We're first going to build this program using the Simple Build Tool (SBT) for Scala. SBT is to Scala what Maven is for Java.

We're first going to build Spark:

sbt/sbt -Dhadoop.version=2.6.0 assembly

Notice the hadoop version I'm passing in.  You'll want to change this to the version of hadoop running on your existing cluster.

The script took nearly 20 minutes to run and completed with this line:

[success] Total time: 1039 s, completed Feb 26, 2015 4:42:04 PM

SparkPi is a sample application that computes an approximate value for pi:

./bin/run-example SparkPi 10

The end of a successful run looks like this:

15/02/26 16:59:29 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool   
15/02/26 16:59:29 INFO DAGScheduler: Job 0 finished: reduce at SparkPi.scala:35, took 6.157368 s  
Pi is roughly 3.14192  
15/02/26 16:59:29 INFO SparkUI: Stopped Spark web UI at http://192.168.1.10:4040  
15/02/26 16:59:29 INFO DAGScheduler: Stopping DAGScheduler  
15/02/26 16:59:30 INFO MapOutputTrackerMasterActor: MapOutputTrackerActor stopped!  
15/02/26 16:59:30 INFO MemoryStore: MemoryStore cleared  
15/02/26 16:59:30 INFO BlockManager: BlockManager stopped  
15/02/26 16:59:30 INFO BlockManagerMaster: BlockManagerMaster stopped  
15/02/26 16:59:30 INFO SparkContext: Successfully stopped SparkContext  
15/02/26 16:59:30 INFO RemoteActorRefProvider$RemotingTerminator: Shutting down remote daemon.  
15/02/26 16:59:30 INFO RemoteActorRefProvider$RemotingTerminator: Remote daemon shut down; proceeding with flushing remote transports.  
15/02/26 16:59:30 INFO RemoteActorRefProvider$RemotingTerminator: Remoting shut down.  

I'm using this code to perform the WordCount program on an existing dataset on my HDFS cluster and output the results:

var file = sc.textFile("hdfs://192.168.1.70:9000/nyt/singles")
val count = file.flatMap(line => line.split(",")).map(word => (word, 1)).reduceByKey(_+_)
count.saveAsTextFile("hdfs://192.168.1.70:9000/nyt/out/02")

You can determine the port number of your namenode by typing:

sudo netstat -tulpn

And looking through the output for a port number, typically either 8020 or 9000.

Troubleshooting

1. A Java execution error can occur if the sample application is executed without sudo privileges:

   craig@U14SPARK01:/usr/lib/apache/spark/1.2.1$ ./bin/run-example SparkPi 10
   /usr/lib/apache/spark/1.2.1/bin/spark-class: line 114: [: : integer expression expected
   /usr/lib/apache/spark/1.2.1/bin/spark-class: line 188: /usr/lib/jvm/jdk/1.0.8_31/bin/java: No such file or directory
   

References

1. Installing Apache Spark 1.1.0 on Ubuntu 14.04
2. Troubleshooting
1. [StackOverflow] Can't execute javac when running sbt

Installing Scala on Ubuntu

Environment

1. Scala 2.11.6
2. Ubuntu 14.10

1. Installing Scala

The following commands have been tested, and are operational, under Ubuntu 14.10

1.1 Downloading the Latest Version

The latest version can be downloaded from the Scala site:

sudo wget http://www.scala-lang.org/files/archive/scala-2.11.6.tgz

The commands will download the scala archive to your current working directory.

1.2 Copy and Untar the Archive

I recommend copying the archive into this folder:

/usr/local/src/scala

Untar the file using this command:

sudo tar -zxvf scala-2.11.6.tgz

2. Modifying the Path

Modify both the PATH and CLASSPATH to point to the new Scala installation.

I like to use gedit to edit my environment file:

sudo gedit /etc/environment

The text in bold was added:

PATH="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/usr/local/src/scala/scala-2.11.6/bin"  
...  
export SCALA_HOME=/usr/local/src/scala/scala-2.11.6

Once the environment file is saved, reload it:

source /etc/environment

3. Verifying the Installation

If the installation and environment editing were both successful, you should be able to find the version of scala on the terminal session:

craig@U14BASEBIG01:/usr/local/src/scala/scala-2.10.4$ scala -version
Scala code runner version 2.11.6 -- Copyright 2002-2013, LAMP/EPFL

You should also be able to enter the scala interactive shell by just typing scala:

craig@U14BASEBIG01:/usr/local/src/scala/scala-2.10.4$ scala
Welcome to Scala version 2.11.6 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_31).  
Type in expressions to have them evaluated.  
Type :help for more information.  
scala>   

References

1. [Blog] Installing Apache Spark on 14.04