!pyspark

"""
Explanation: This notebook was prepared by Donne Martin. Source and license info is on GitHub.
Spark

IPython Notebook Setup
Python Shell
DataFrames
RDDs
Pair RDDs
Running Spark on a Cluster
Viewing the Spark Application UI
Working with Partitions
Caching RDDs
Checkpointing RDDs
Writing and Running a Spark Application
Configuring Spark Applications
Streaming
Streaming with States
Broadcast Variables
Accumulators

IPython Notebook Setup
The dev-setup repo contains scripts to install Spark and to automate the its integration with IPython Notebook through the pydata.sh script.
You can also follow the instructions provided here to configure IPython Notebook Support for PySpark with Python 2.
To run Python 3 with Spark 1.4+, check out the following posts on Stack Overflow or Reddit.
Python Shell
Start the pyspark shell (REPL):
End of explanation
"""


sc

"""
Explanation: View the spark context, the main entry point to the Spark API:
End of explanation
"""


users = context.load("s3n://path/to/users.json", "json")

"""
Explanation: DataFrames
From the following reference:
A DataFrame is a distributed collection of data organized into named columns. It is conceptually equivalent to a table in a relational database or a data frame in R/Python, but with richer optimizations under the hood. 
Create a DataFrame from JSON files on S3:
End of explanation
"""


young = users.filter(users.age<21)

"""
Explanation: Create a new DataFrame that contains “young users” only:
End of explanation
"""


young = users[users.age<21]

"""
Explanation: Alternatively, using Pandas-like syntax:
End of explanation
"""


young.select(young.name, young.age+1)

"""
Explanation: Increment everybody’s age by 1:
End of explanation
"""


young.groupBy("gender").count()

"""
Explanation: Count the number of young users by gender:
End of explanation
"""


young.join(logs, logs.userId == users.userId, "left_outer")

"""
Explanation: Join young users with another DataFrame called logs:
End of explanation
"""


young.registerTempTable("young")
context.sql("SELECT count(*) FROM young")

"""
Explanation: Count the number of users in the young DataFrame:
End of explanation
"""


pandas_df = young.toPandas()

"""
Explanation: Convert Spark DataFrame to Pandas:
End of explanation
"""


spark_df = context.createDataFrame(pandas_df)

"""
Explanation: Create a Spark DataFrame from Pandas:
End of explanation
"""


from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

"""
Explanation: Given the Spark Context, create a SQLContext:
End of explanation
"""


df = sqlContext.jsonFile("file:/path/file.json")

"""
Explanation: Create a DataFrame based on the content of a file:
End of explanation
"""


df.show()

"""
Explanation: Display the content of the DataFrame:
End of explanation
"""


df.printSchema()

"""
Explanation: Print the schema:
End of explanation
"""


df.select("column_name")

"""
Explanation: Select a column:
End of explanation
"""


df.filter(df.column_name>10)

"""
Explanation: Create a DataFrame with rows matching a given filter:
End of explanation
"""


df.groupBy("column_name").count()

"""
Explanation: Aggregate the results and count:
End of explanation
"""


df = sqlContext.inferSchema(my_data)

"""
Explanation: Convert a RDD to a DataFrame (by inferring the schema):
End of explanation
"""


df.registerTempTable("dataframe_name")

"""
Explanation: Register the DataFrame as a table:
End of explanation
"""


rdd_from_df = sqlContext.sql("SELECT * FROM dataframe_name")

"""
Explanation: Run a SQL Query on a DataFrame registered as a table:
End of explanation
"""


my_data = sc.textFile("file:/path/*")

"""
Explanation: RDDs
Note: RDDs are included for completeness.  In Spark 1.3, DataFrames were introduced which are recommended over RDDs.  Check out the DataFrames announcement for more info.
Resilient Distributed Datasets (RDDs) are the fundamental unit of data in Spark.  RDDs can be created from a file, from data in memory, or from another RDD.  RDDs are immutable.
There are two types of RDD operations:
* Actions: Returns values, data is not processed in an RDD until an action is preformed
* Transformations: Defines a new RDD based on the current
Create an RDD from the contents of a directory:
End of explanation
"""


my_data.count()

"""
Explanation: Count the number of lines in the data:
End of explanation
"""


my_data.collect()

"""
Explanation: Return all the elements of the dataset as an array--this is usually more useful after a filter or other operation that returns a sufficiently small subset of the data:
End of explanation
"""


my_data.take(10)

"""
Explanation: Return the first 10 lines in the data:
End of explanation
"""


my_data.filter(lambda line: ".txt" in line)

"""
Explanation: Create an RDD with lines matching the given filter:
End of explanation
"""


sc.textFile("file:/path/file.txt") \
    .filter(lambda line: ".txt" in line) \
    .count()

"""
Explanation: Chain a series of commands:
End of explanation
"""


first_words = my_data.map(lambda line: line.split()[0])

"""
Explanation: Create a new RDD mapping each line to an array of words, taking only the first word of each array:
End of explanation
"""


for word in first_words.take(10):
    print word

"""
Explanation: Output each word in first_words:
End of explanation
"""


first_words.saveAsTextFile("file:/path/file")

"""
Explanation: Save the first words to a text file:
End of explanation
"""


DATE_TIME = 0
USER_ID = 1
IP_ADDRESS = 2
ACTION = 3

log_data = sc.textFile("file:/path/*")

user_actions = log_data \
    .map(lambda line: line.split()) \
    .map(lambda words: (words[USER_ID], 1))  \
    .reduceByKey(lambda count1, count2: count1 + count2)

"""
Explanation: Pair RDDs
Pair RDDs contain elements that are key-value pairs.  Keys and values can be any type.
Given a log file with the following space deilmited format: [date_time, user_id, ip_address, action], map each request to (user_id, 1):
End of explanation
"""


user_actions.map(lambda pair: (pair[0], pair[1])).sortyByKey(False).take(5)

"""
Explanation: Show the top 5 users by count, sorted in descending order:
End of explanation
"""


user_ips = log_data \
    .map(lambda line: line.split()) \
    .map(lambda words: (words[IP_ADDRESS],words[USER_ID])) \
    .groupByKey()

"""
Explanation: Group IP addresses by user id:
End of explanation
"""


user_data = sc.textFile("file:/path/*")

user_profile = user_data \
    .map(lambda line: line.split(',')) \
    .map(lambda words: (words[0], words[1:]))

"""
Explanation: Given a user table with the following csv format: [user_id, user_info0, user_info1, ...], map each line to (user_id, [user_info...]):
End of explanation
"""


user_actions_with_profile = user_actions.join(user_profile)

"""
Explanation: Inner join the user_actions and user_profile RDDs:
End of explanation
"""


for (user_id, (user_info, count)) in user_actions_with_profiles.take(10):
    print user_id, count, user_info

"""
Explanation: Show the joined table:
End of explanation
"""


!sudo service spark-master start
!sudo service spark-worker start

"""
Explanation: Running Spark on a Cluster
Start the standalone cluster's Master and Worker daemons:
End of explanation
"""


!sudo service spark-master stop
!sudo service spark-worker stop

"""
Explanation: Stop the standalone cluster's Master and Worker daemons:
End of explanation
"""


!sudo service spark-master stop
!sudo service spark-worker stop

"""
Explanation: Restart the standalone cluster's Master and Worker daemons:
End of explanation
"""


http://localhost:18080//

"""
Explanation: View the Spark standalone cluster UI:
End of explanation
"""


!MASTER=spark://localhost:7077 pyspark

"""
Explanation: Start the Spark shell and connect to the cluster:
End of explanation
"""


sc.master

"""
Explanation: Confirm you are connected to the correct master:
End of explanation
"""


http://localhost:4040/

"""
Explanation: Viewing the Spark Application UI
From the following reference:
Every SparkContext launches a web UI, by default on port 4040, that displays useful information about the application. This includes:
A list of scheduler stages and tasks
A summary of RDD sizes and memory usage
Environmental information.
Information about the running executors
You can access this interface by simply opening http://<driver-node>:4040 in a web browser. If multiple SparkContexts are running on the same host, they will bind to successive ports beginning with 4040 (4041, 4042, etc).
Note that this information is only available for the duration of the application by default. To view the web UI after the fact, set spark.eventLog.enabled to true before starting the application. This configures Spark to log Spark events that encode the information displayed in the UI to persisted storage.
End of explanation
"""


val dbConnection = ...
lines.map(... dbConnection.createStatement(...) ...)
dbConnection.close() // Wrong!

"""
Explanation: Working with Partitions
From the following reference:
The Spark map() and flatMap() methods only operate on one input at a time, and provide no means to execute code before or after transforming a batch of values. It looks possible to simply put the setup and cleanup code before and after a call to map() in Spark:
End of explanation
"""


def count_txt(partIter):
    for line in partIter: 
        if ".txt" in line: txt_count += 1
    yield (txt_count)

my_data = sc.textFile("file:/path/*") \
    .mapPartitions(count_txt) \
    .collect()
    
# Show the partitioning 
print "Data partitions: ", my_data.toDebugString()

"""
Explanation: However, this fails for several reasons:

It puts the object dbConnection into the map function’s closure, which requires that it be serializable (for example, by implementing java.io.Serializable). An object like a database connection is generally not serializable.
map() is a transformation, rather than an operation, and is lazily evaluated. The connection can’t be closed immediately here.
Even so, it would only close the connection on the driver, not necessarily freeing resources allocated by serialized copies.

In fact, neither map() nor flatMap() is the closest counterpart to a Mapper in Spark — it’s the important mapPartitions() method. This method does not map just one value to one other value, but rather maps an Iterator of values to an Iterator of other values. It’s like a “bulk map” method. This means that the mapPartitions() function can allocate resources locally at its start, and release them when done mapping many values.
End of explanation
"""


# Cache RDD to memory
my_data.cache()

# Persist RDD to both memory and disk (if memory is not enough), with replication of 2
my_data.persist(MEMORY_AND_DISK_2)

# Unpersist RDD, removing it from memory and disk
my_data.unpersist()

# Change the persistence level after unpersist
my_data.persist(MEMORY_AND_DISK)

"""
Explanation: Caching RDDs
Caching an RDD saves the data in memory.  Caching is a suggestion to Spark as it is memory dependent.
By default, every RDD operation executes the entire lineage.  Caching can boost performance for datasets that are likely to be used by saving this expensive recomputation and is ideal for iterative algorithms or machine learning.

cache() stores data in memory
persist() stores data in MEMORY_ONLY, MEMORY_AND_DISK (spill to disk), and  DISK_ONLY

Disk memory is stored on the node, not on HDFS.
Replication is possible by using MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.  If a cached partition becomes unavailable, Spark recomputes the partition through the lineage.
Serialization is possible with MEMORY_ONLY_SER and MEMORY_AND_DISK_SER.  This is more space efficient but less time efficient, as it uses Java serialization by default.
End of explanation
"""


# Enable checkpointing by setting the checkpoint directory, 
# which will contain all checkpoints for the given data:
sc.setCheckpointDir("checkpoints")

my_data = sc.parallelize([1,2,3,4,5])

# Long loop that may cause a stack overflow
for i in range(1000):
    my_data = mydata.map(lambda myInt: myInt + 1)

    if i % 10 == 0: 
        my_data.checkpoint()
        my_data.count()

my_data.collect()
     
# Display the lineage
for rddstring in my_data.toDebugString().split('\n'): 
    print rddstring.strip()

"""
Explanation: Checkpointing RDDs
Caching maintains RDD lineage, providing resilience.  If the lineage is very long, it is possible to get a stack overflow.
Checkpointing saves the data to HDFS, which provide fault tolerant storage across nodes.  HDFS is not as fast as local storage for both reading and writing.  Checkpointing is good for long lineages and for very large data sets that might not fit on local storage.  Checkpointing removes lineage.
Create a checkpoint and perform an action by calling count() to materialize the checkpoint and save it to the checkpoint file:
End of explanation
"""


import sys
from pyspark import SparkContext

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print >> sys.stderr, "Usage: App Name <file>"
        exit(-1)
        
    count_text_files()
    
def count_text_files():
    sc = SparkContext()
    logfile = sys.argv[1]
    text_files_count = sc.textFile(logfile)
        .filter(lambda line: '.txt' in line)
    text_files_count.cache()
    print("Number of text files: ", text_files_count.count())

"""
Explanation: Writing and Running a Spark Application
Create a Spark application to count the number of text files:
End of explanation
"""


!spark-submit --properties-file dir/myspark.conf script.py data/*

"""
Explanation: Submit the script to Spark for processing:
End of explanation
"""


!spark-submit --master spark//localhost:7077 --name 'App Name' script.py data/*

"""
Explanation: Configuring Spark Applications
Run a Spark app and set the configuration options in the command line:
End of explanation
"""


spark.app.name  App Name
spark.ui.port   4141
spark.master    spark://localhost:7077

"""
Explanation: Configure spark.conf:
End of explanation
"""


!spark-submit --properties-file spark.conf script.py data/*

"""
Explanation: Run a Spark app and set the configuration options through spark.conf:
End of explanation
"""


sconf = SparkConf() \
    .setAppName("Word Count") \
    .set("spark.ui.port","4141")
sc = SparkContext(conf=sconf)

"""
Explanation: Set the config options programmatically:
End of explanation
"""


$SPARK_HOME/conf/log4j.properties.template

"""
Explanation: Set logging levels located in the following file, or place a copy in your pwd:
End of explanation
"""


!spark-shell --master local[2]

"""
Explanation: Streaming
Start the Spark Shell locally with at least two threads (need a minimum of two threads for streaming, one for receiving, one for processing):
End of explanation
"""


val ssc = new StreamingContext(new SparkConf(), Seconds(1)) 
val my_stream = ssc.socketTextStream(hostname, port)

"""
Explanation: Create a StreamingContext (similar to SparkContext in core Spark) with a batch duration of 1 second:
End of explanation
"""


val logs = ssc.socketTextStream(hostname, port)

"""
Explanation: Get a DStream from a streaming data source (text from a socket):
End of explanation
"""


val requests = my_stream
    .map(line => (line.split(" ")(2), 1))
    .reduceByKey((x, y) => x + y)

"""
Explanation: DStreams support regular transformations such as map, flatMap, and filter, and pair transformations such as reduceByKey, groupByKey, and joinByKey.
Apply a DStream operation to each batch of RDDs (count up requests by user id, reduce by key to get the count):
End of explanation
"""


val sorted_requests = requests
    .map(pair => pair.swap)
    .transform(rdd => rdd.sortByKey(false))

"""
Explanation: The transform(function) method creates a new DStream by executing the input function on the RDDs.
End of explanation
"""


sorted_requests.foreachRDD((rdd, time) => {
    println("Top users @ " + time)
    rdd.take(5).foreach(
    pair => printf("User: %s (%s)\n", pair._2, pair._1))
}

"""
Explanation: foreachRDD(function) performs a function on each RDD in the DStream (map is like a shortcut not requiring you to get the RDD first before doing an operation):
End of explanation
"""


requests.saveAsTextFiles("/dir/requests")

"""
Explanation: Save the DStream result part files with the given folder prefix, the actual folder will be /dir/requests-timestamp0/:
End of explanation
"""


ssc.start()

"""
Explanation: Start the execution of all DStreams:
End of explanation
"""


ssc.awaitTermination()

"""
Explanation: Wait for all background threads to complete before ending the main thread:
End of explanation
"""


ssc.checkpoint("dir")

"""
Explanation: Streaming with States
Enable checkpointing to prevent infinite lineages:
End of explanation
"""


def updateCount = (newCounts: Seq[Int], state: Option[Int]) => {
    val newCount = newCounts.foldLeft(0)(_ + _)
    val previousCount = state.getOrElse(0)
    Some(newCount + previousCount)
}

val totalUserreqs = userreqs.updateStateByKey(updateCount)

"""
Explanation: Compute a DStream based on the previous states plus the current state:
End of explanation
"""


val reqcountsByWindow = logs.map(line => (line.split(' ')(2), 1))
    .reduceByKeyAndWindow((x: Int, y: Int) => x + y, Minutes(5), Seconds(30))

"""
Explanation: Compute a DStream based Sliding window, every 30 seconds, count requests by user over the last 5 minutes:
End of explanation
"""


// define listener
class MyListener extends StreamingListener {
  override def onReceiverStopped(...) {
    streamingContext.stop()
  }
} 

// attach listener
streamingContext. addStreamingListener(new MyListener())

"""
Explanation: Collect statistics with the StreamingListener API:
End of explanation
"""


broadcast_file = "broadcast.txt"
broadcast_list = list(map(lambda l: l.strip(), open(broadcast_file)))

"""
Explanation: Broadcast Variables
Read in list of items to broadcast from a local file:
End of explanation
"""


broadcast_list_sc = sc.broadcast(broadcast_list)

"""
Explanation: Broadcast the target list to all workers:
End of explanation
"""


log_file = "hdfs://localhost/user/logs/*"
filtered_data = sc.textFile(log_file)\
    .filter(lambda line: any(item in line for item in broadcast_list_sc.value))

filtered_data.take(10)

"""
Explanation: Filter based on the broadcast list:
End of explanation
"""


txt_count = sc.accumulator(0)

"""
Explanation: Accumulators
Create an accumulator:
End of explanation
"""


my_data = sc.textFile(filePath)
my_data.foreach(lambda line: if '.txt' in line: txt_count.add(1))

"""
Explanation: Count the number of txt files in the RDD:
End of explanation
"""


jpg_count = sc.accumulator(0)
html_count = sc.accumulator(0)
css_count = sc.accumulator(0)

def countFileType(s):
    if '.jpg' in s: jpg_count.add(1)
    elif '.html' in s: html_count.add(1)
    elif '.css' in s: css_count.add(1)

filename="hdfs://logs/*"

logs = sc.textFile(filename)
logs.foreach(lambda line: countFileType(line))

print  'File Type Totals:'
print '.css files: ', css_count.value
print '.html files: ', html_count.value
print '.jpg files: ', jpg_count.value

"""
Explanation: Count the number of file types encountered:
End of explanation
"""


