sudo -u spark vi /usr/hdp/current/spark-client/conf/spark-env.sh

export HADOOP_CONF_DIR=/etc/hadoop/2.4.2.0-258/0/
export SPARK_LOCAL_IP=0.0.0.0
export SPARK_PUBLIC_DNS="current_node_public_dns"
export SPARK_WORKER_CORES=6
export SPARK_MASTER_WEBUI_PORT=8081

export LD_LIBRARY_PATH=/usr/hdp/current/hadoop-client/lib/native/:$LD_LIBRARY_PATH

spark_worker1_public_dns  
spark_worker2_public_dns  
spark_worker3_public_dns  

sudo su hdfs
HDP_VERSION=2.4.2.0-258
SPARK_JAR=spark-assembly-1.6.1.2.4.2.0-258-hadoop2.7.1.2.4.2.0-258.jar
hdfs dfs -mkdir "/hdp/apps/${HDP_VERSION}/spark/"
hdfs dfs -put "/usr/hdp/${HDP_VERSION}/spark/lib/$SPARK_JAR" "/hdp/apps/${HDP_VERSION}/spark/spark-hdp-assembly.jar"

No spark assembly jar for HDP on HDFS, defaultSparkAssembly:hdfs://hostname:8020/hdp/apps/2.4.2.0-258/spark/spark-hdp-assembly.jar

cd /usr/hdp/current/spark-client
sudo -u spark sbin/start-master.sh -h 0.0.0.0 -p 7077

sudo -u spark mkdir /var/run/spark/work
cd /usr/hdp/current/spark-client
sudo -u spark sbin/start-slave.sh spark://localhost:7077

sudo -u spark vi conf/spark-env.sh
export SPARK_LOCAL_IP=127.0.0.1

cd /usr/hdp/current/spark-client
sudo -u spark sbin/stop-master.sh 
sudo -u spark sbin/stop-slave.sh

cd /usr/hdp/current/spark-client
sudo -u spark spark-submit --class org.apache.spark.examples.SparkPi --master spark://hostname:7077 --num-executors 3 --driver-memory 512m --executor-memory 512m --executor-cores 1 lib/spark-examples*.jar 10

./bin/run-example SparkPi

cd $SPARK_HOME
export HADOOP_USER_NAME=spark
spark-submit --class org.apache.spark.examples.SparkPi --master spark://sf-master:7077 --num-executors 3 --driver-memory 512m --executor-memory 512m --executor-cores 1 examples/jars/spark-examples*.jar 10

package spark.pi

import scala.math.random
import org.apache.spark._

/** Computes an approximation to pi */
object SparkPi {
  def main(args: Array[String]) {
    val conf = new SparkConf()
      .setAppName("Spark Pi From Anderson")
      .setMaster("spark://localhost:7077")
      .set("spark.executor.memory", "512m")
      .set("spark.cores.max", "3")

    val spark = new SparkContext(conf)
//    spark.addJar("target/scala-2.11/scalatutorial_2.11-1.0.jar")

    val slices = if (args.length > 0) args(0).toInt else 2
    val n = 100000 * slices
    val count = spark.parallelize(1 to n, slices).map { i =>
      val x = random * 2 - 1
      val y = random * 2 - 1
      if (x * x + y * y < 1) 1 else 0
    }.reduce(_ + _)
    println("Pi is roughly " + 4.0 * count / n)
    spark.stop()
  }
}
 

spark.addJar("target/scala-2.11/scalatutorial_2.11-1.0.jar")

sbt package

HADOOP_USER_NAME=spark /home/anderson/apps/spark-1.6.1-bin-hadoop2.6/bin/spark-submit --class spark.pi.SparkPi  --master spark://hostname:7077 target/scala-2.10/scalatutorial_2.10-1.0.jar 10

cd  /usr/hdp/current/spark-client
sudo -u spark spark-submit --class org.apache.spark.examples.SparkPi --master yarn --num-executors 3 --driver-memory 512m --executor-memory 512m --executor-cores 1 lib/spark-examples*.jar 10

mkdir -p ~/apps/hdp_conf
scp -i ~/.ssh/dev.pem ubuntu@yarn-master:/etc/hadoop/2.4.2.0-258/0/yarn-site.xml ~/apps/hdp_conf/
scp -i ~/.ssh/dev.pem ubuntu@yarn-master:/etc/hadoop/2.4.2.0-258/0/core-site.xml ~/apps/hdp_conf/
export HADOOP_CONF_DIR=/home/anderson/apps/hdp_conf/
export HADOOP_USER_NAME=spark

spark-submit --class org.apache.spark.examples.SparkPi --master yarn --num-executors 3 --driver-memory 512m --executor-memory 512m --executor-cores 1 $SPARK_HOME/examples/jars/spark-examples*.jar 10

echo "deb https://dl.bintray.com/sbt/debian /" | sudo tee -a /etc/apt/sources.list.d/sbt.list
sudo apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 642AC823
sudo apt-get update
sudo apt-get install sbt

name := "ScalaTutorial"
version := "1.0"
scalaVersion := "2.10.5"
libraryDependencies += "org.apache.spark" % "spark-core_2.11" % "1.6.2"

name := "scala-lab"
version := "1.0"
scalaVersion := "2.10.5"

// https://mvnrepository.com/artifact/org.apache.spark/spark-core_2.10
libraryDependencies += "org.apache.spark" % "spark-core_2.11" % "2.0.0"

import org.apache.spark.{SparkConf, SparkContext}

object SimpleApp {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    conf.setAppName("Anderson's Simple Application")
    conf.setMaster("local[*]")

    val sc = new SparkContext(conf)
    val data = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val rdd = sc.parallelize(data)
    println(rdd.reduce((a, b) => a + b))
  }
}

Exception in thread "main" java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfig
	at org.apache.hadoop.yarn.client.api.TimelineClient.createTimelineClient(TimelineClient.java:55)
	at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.createTimelineClient(YarnClientImpl.java:181)
	at org.apache.hadoop.yarn.client.api.impl.YarnClientImpl.serviceInit(YarnClientImpl.java:168)
	at org.apache.hadoop.service.AbstractService.init(AbstractService.java:163)
	at org.apache.spark.deploy.yarn.Client.submitApplication(Client.scala:150)
	at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend.start(YarnClientSchedulerBackend.scala:56)
	at org.apache.spark.scheduler.TaskSchedulerImpl.start(TaskSchedulerImpl.scala:149)
	at org.apache.spark.SparkContext.<init>(SparkContext.scala:500)
	at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2256)
	at org.apache.spark.sql.SparkSession$Builder$$anonfun$8.apply(SparkSession.scala:831)
	at org.apache.spark.sql.SparkSession$Builder$$anonfun$8.apply(SparkSession.scala:823)
	at scala.Option.getOrElse(Option.scala:121)
	at org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:823)
	at org.apache.spark.examples.SparkPi$.main(SparkPi.scala:31)
	at org.apache.spark.examples.SparkPi.main(SparkPi.scala)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:729)
	at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:185)
	at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:210)
	at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:124)
	at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: com.sun.jersey.api.client.config.ClientConfig
	at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
	... 24 more

sudo -u spark  vi conf/spark-env.sh

SPARK_LOCAL_IP=127.0.0.1

export SPARK_HOME=/home/anderson/apps/spark-1.6.2-bin-hadoop2.6
export PYTHONPATH=$PYTHONPATH:$SPARK_HOME/python
export PYTHONPATH=$PYTHONPATH:$SPARK_HOME/python/lib/py4j-0.9-src.zip

export CLASSPATH=$CLASSPATH:/path/postgresql-9.3-1103.jdbc3.jar
export SPARK_CLASSPATH=/path/postgresql-9.3-1103.jdbc3.jar

import pyspark

from pyspark import SparkContext, SparkConf

spconf = SparkConf().setMaster('local').setAppName('Tutorial')
sc = SparkContext(conf=spconf)

spark-submit tutorial.py

textFile = sc.textFile('README.md')
print textFile.count()  # 36 ;the number of lines
print textFile.first()  # Examples for Learning Spark

pythonLines = textFile.filter(lambda line: 'python' in line.lower())
print pythonLines.first()

textFile = sc.textFile('README.md')
def max(a, b):
    return a if a > b else b

textFile.map(lambda line: len(line.split(' '))).reduce(max) # 36

wordCounts = textFile.flatMap(lambda line: line.split()).\
    map(lambda word: (word, 1)).reduceByKey(combine)
print sorted(wordCounts.collect(), key=lambda d: d[1], reverse=True)
# [(u'the', 10), (u'*', 6), (u'Spark', 5), (u'for', 4), (u'of', 4), ...]

rdd.cache()

from pyspark import StorageLevel
...
rdd.persist(StorageLevel.MEMORY_ONLY_SER_2)

data = sc.parallelize(['a', 'b', 'c', 'd', 'e', 1,2,3,4,5])
data.cache()

alpha = data.filter(lambda word: type(word) == str or type(word) == unicode)
alpha.cache()

numeric = data.filter(lambda word: type(word) == int)
numeric.cache()

print alpha.collect() #  ['a', 'b', 'c', 'd', 'e']
print numeric.collect() # [1, 2, 3, 4, 5]

print alpha.union(numeric).collect() # ['a', 'b', 'c', 'd', 'e', 1, 2, 3, 4, 5]

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10,11,12,13,14,15])
for line in rdd.take(5):
    print line # 1 2 3 4 5

rdd = sc.parallelize(['a', 'b', 'c', 'd', 'e', 'z', 'y', 'x', 'f', 'g'])
rdd.top(3) # ['z', 'y', 'x']

rdd1 = sc.parallelize([1,2,3,4,5, 'a'])
rdd2 = sc.parallelize(['a', 'b', 'c', 1, 2])

rdd1.union(rdd2).collect() # [1, 2, 3, 4, 5, 'a', 'a', 'b', 'c', 1, 2]
rdd1.intersection(rdd2).collect() # ['a', 2, 1]
rdd1.subtract(rdd2).collect() # [4, 3, 5]

rdd1 = sc.parallelize([1,2])
rdd2 = sc.parallelize(['a', 'b', 'c'])

rdd1.cartesian(rdd2).collect()
# [(1, 'a'), (1, 'b'), (1, 'c'), (2, 'a'), (2, 'b'), (2, 'c')]

rdd = sc.parallelize([1,1,2,2,2,3,4])
rdd.distinct().collect() # [1, 2, 3, 4]

rdd = sc.parallelize([10,20,30,40,50,60,70,80,90,100])
rdd.sample(False, 0.5).collect()
# [40, 50, 60]
# [30, 40, 50, 90, 100]
# [50, 80, 90, 100]

rdd.sample(True, 0.5).collect()
# [10, 10, 20, 30, 60, 60, 80, 90, 100, 100]
# [10, 100, 100]

rdd.sample(True, 0).collect()
# []

rdd = sc.parallelize([1,2,3,4,5,6,7,8,9,10])
sum, avg = rdd.aggregate((0, 0),
              lambda acc, v: (acc[0] + v, acc[1] + 1),
              lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1]))
# sum: 55, avg: 10
print sum/float(avg) # 5.5

export CLASSPATH=$CLASSPATH:/path/postgresql-9.3-1103.jdbc3.jar
export SPARK_CLASSPATH=/path/postgresql-9.3-1103.jdbc3.jar

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext

spconf = SparkConf().setMaster('local').setAppName('Tutorial')
sc = SparkContext(conf=spconf)
sqlContext = SQLContext(sc)

df = sqlContext.read.json("examples/src/main/resources/people.json")
df.show()
# +----+-------+
# | age|   name|
# +----+-------+
# |null|Michael|
# |  30|   Andy|
# |  19| Justin|
# +----+-------+

df.printSchema()
# root
#  |-- age: long (nullable = true)
#  |-- name: string (nullable = true)

df.select("name").show()
# +-------+
# |   name|
# +-------+
# |Michael|
# |   Andy|
# | Justin|
# +-------+

df.select(df['name'], df['age'] + 1).show()
# +-------+---------+
# |   name|(age + 1)|
# +-------+---------+
# |Michael|     null|
# |   Andy|       31|
# | Justin|       20|
# +-------+---------+

df.filter(df['age'] > 21).show()
# +---+----+
# |age|name|
# +---+----+
# | 30|Andy|
# +---+----+

df.groupBy("age").count().show()
# +----+-----+
# | age|count|
# +----+-----+
# |null|    1|
# |  19|    1|
# |  30|    1|
# +----+-----+

import os
os.environ['SPARK_CLASSPATH'] = "/path/to/postgresql-9.3-1103.jdbc3.jar"

df = sqlContext.read.format("jdbc").options(
    url="jdbc:postgresql://host:port/database",
    port=5432,
    driver = "org.postgresql.Driver",
    dbtable = "users",
    user="user_name",
    password="password").load()

sqlContext.registerDataFrameAsTable(df, 'users')
df2 = sqlContext.sql('select id, nickname from users where id=155 limit 1')
print df2.collect()
# [Row(id=155, nickname=u'\uc575\uadf8\ub9ac\uc564\ub354\uc2a8')]