In this repo, I try to use Spark (PySpark) to look into a downloading log file in
.CSV format. This repo can be considered as an introduction to the very basic functions of Spark. It may be helpful for those who are beginners to Spark.
- Hadoop knowledge will not be covered in this practice.
- I used single-node mode here. Cluster deployment will not be discussed.
- If you're also interested in using Scala for Spark rather than Python, you can refer to my another GitHub repo spark-practice-scala.
Additionally, we're using a real log file as sample data in this tutorial and trying to cover some operations commonly used in daily works. If you would like to get to know more operations with minimal sample data, you can refer to a seperate script I prepared, Basic Operations in PySpark.
- 1. Preparation
- 2. Sample Data
- 3. How We Use Spark (PySpark) Interactively
- 4. Submitting Application
- 5. Spark SQL and DataFrames
The environment I worked on is an Ubuntu machine. It's quite simple to install Spark on Ubuntu platform.
Firstly, ensure that JAVA is install properly. If not, we can install by
$ sudo apt-get install openjdk-8-jdk
Then we can download the latest version of Spark from http://spark.apache.org/downloads.html and unzip it. Then we can simply test if Spark runs properly by running the command below in the Spark directory
2. Sample Data
The sample data we use is from http://cran-logs.rstudio.com/. It is the full downloads log of R packages from Rstudio's CRAN mirror on December 12 2015 (you can get the data in the
sample_data folder of this repository).
We will try to use Spark to do some simple analytics on this data.
3. How We Use Spark (PySpark) Interactively
We can directly call
pyspark to start Spark
Instead, we can also use IPython. It can bring some convenient features like auto-completion.
$ PYSPARK_DRIVER_PYTHON=ipython ./bin/pyspark
If you have Jupyter installed, you can also choose to run PySpark in Jupyter Notebook or Jupyter QtConsole.
$ PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=notebook ./bin/pyspark $ PYSPARK_DRIVER_PYTHON=jupyter PYSPARK_DRIVER_PYTHON_OPTS=qtconsole ./bin/pyspark
If you want to specify the Python version you prefer, you can use
$ PYSPARK_PYTHON=python3.6 ./bin/pyspark
After Spark is started, a default SparkContext will be created (usually named as "sc").
The most common method used to load data is
sc.textFile. This method takes an URI for the file (local file or other URI like
hdfs://), and will read the data as a collections of lines.
# Load the data >>> raw_content = sc.textFile("2015-12-12.csv") # Print the type of the object >>> type(raw_content) <class 'pyspark.rdd.RDD'> # Print the number of lines >>> raw_content.count() 421970
You may want to note that all Spark’s file-based input methods, including
textFile, support running on directories, compressed files, and wildcards as well . For example, you can use
textFile("/my/directory/*.gz"). In our case, the two commands below will help load exactly the same data.
>>> a = sc.textFile("2015-12-12.csv") >>> b = sc.textFile("2015-12-12.csv.gz") >>> a.count() 421970 >>> b.count() 421970
This feature makes things much simpler when we have multiple text data files to load. By giving the directory under where these files are, we can load many data files with only one line. Additionally, we can also use wildcards to specify the file types we would like to load, like with
textFile("/my/directory/*.txt"), we will only load those files with
.txt file type in the directory we specified.
Show the Head (First
We can use
take method to return first
>>> raw_content.take(5) [u'"date","time","size","r_version","r_arch","r_os","package","version","country","ip_id"', u'"2015-12-12","13:42:10",257886,"3.2.2","i386","mingw32","HistData","0.7-6","CZ",1', u'"2015-12-12","13:24:37",1236751,"3.2.2","x86_64","mingw32","RJSONIO","1.3-0","DE",2', u'"2015-12-12","13:42:35",2077876,"3.2.2","i386","mingw32","UsingR","2.0-5","CZ",1', u'"2015-12-12","13:42:01",266724,"3.2.2","i386","mingw32","gridExtra","2.0.0","CZ",1']
We can also take samples randomly with
takeSample method. With
takeSample method, we can give three arguments and need to give at least two of them. They are
number of samples, and
>>> raw_content.takeSample(True, 5, 3) [u'"2015-12-12","16:41:22",18773,"3.2.3","x86_64","mingw32","evaluate","0.8","US",10935', u'"2015-12-12","13:06:32",494138,"3.2.3","x86_64","linux-gnu","rjson","0.2.15","KR",655', u'"2015-12-12","03:50:05",140207,NA,NA,NA,"SACOBRA","0.7","DE",129', u'"2015-12-12","21:40:13",622505,"3.2.3","x86_64","linux-gnu","stratification","2.2-5","US",4860', u'"2015-12-12","23:52:06",805204,"3.2.2","x86_64","mingw32","readxl","0.1.0","CA",104']
If we specified the last argument, i.e.
seed, we would be able to reproduce the samples exactly.
Transformation (map & flatMap)
We may need to note that each row of the data is a character string, and it would be more convenient to have an array in some senarios. We can use
map to transform them and use
take method to get the first a few rows to check how the resutls look like.
>>> content = raw_content.map(lambda x: x.split(',')) >>> content.take(3) [ [u'"date"', u'"time"', u'"size"', u'"r_version"', u'"r_arch"', u'"r_os"', u'"package"', u'"version"', u'"country"', u'"ip_id"'], [u'"2015-12-12"', u'"13:42:10"', u'257886', u'"3.2.2"', u'"i386"', u'"mingw32"', u'"HistData"', u'"0.7-6"', u'"CZ"', u'1'], [u'"2015-12-12"', u'"13:24:37"', u'1236751', u'"3.2.2"', u'"x86_64"', u'"mingw32"', u'"RJSONIO"', u'"1.3-0"', u'"DE"', u'2'] ]
map(function) method is one of the most basic and important methods in Spark. It returns a new distributed dataset formed by passing each element of the source through a function specified by user .
There are several ways to define the functions for
map. Normally, we can use lambda function to do this, just like what I did above. This is suitable for simple functions (one line statement). For more complicated process, we can also define a separate function in Python fashion and invoke it within
We have an example here: you may have noted the double quotation marks in the imported data above, and I want to remove all of them in each element of our data
# remove the double quotation marks in the imported data >>> def clean(x): return([xx.replace('"', '') for xx in x]) >>> content = content.map(clean) >>> content.take(4) [ [u'date', u'time', u'size', u'r_version', u'r_arch', u'r_os', u'package', u'version', u'country', u'ip_id'], [u'2015-12-12', u'13:42:10', u'257886', u'3.2.2', u'i386', u'mingw32', u'HistData', u'0.7-6', u'CZ', u'1'], [u'2015-12-12', u'13:24:37', u'1236751', u'3.2.2', u'x86_64', u'mingw32', u'RJSONIO', u'1.3-0', u'DE', u'2'], [u'2015-12-12', u'13:42:35', u'2077876', u'3.2.2', u'i386', u'mingw32', u'UsingR', u'2.0-5', u'CZ', u'1'] ]
We can also use multiple
map operators in a single statement. For example,
raw_content.map(lambda x: x.split(',')).map(clean) will return the same results.
The same function-defining approach is also applicable to
filter method which will be introduced later.
You may have noted that there is another method named
flatMap. Then what's the difference between
flatMap? We can look into a simple example firstly.
>>> text=["a b c", "d e", "f g h"] >>> sc.parallelize(text).map(lambda x:x.split(" ")).collect() [['a', 'b', 'c'], ['d', 'e'], ['f', 'g', 'h']] >>> sc.parallelize(text).flatMap(lambda x:x.split(" ")).collect() ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h']
To put it simple (maybe not precise), we can say that
map will return a sequence of the same length as the original data. In this sequence each element is a sub-sequence corresponding to one element in original data.
flatMap will return a sequence whose length equals to the sum of the lengths of all sub-sequance returned by
Reduce and Counting
I would like to know how many downloading records each package has. For example, for R package "Rcpp", I want to know how many rows belong to it.
>>> # Note here x is just the 7th element of each row, that is the package name. >>> package_count = content.map(lambda x: (x, 1)).reduceByKey(lambda a,b: a+b) >>> type(package_count) <class 'pyspark.rdd.PipelinedRDD'> >>> package_count.count() 8660 >>> package_count.take(5) [(u'SIS', 24), (u'StatMethRank', 15), (u'dbmss', 54), (u'searchable', 14), (u'RcmdrPlugin.TextMining', 3)]
To achive the same purpose, we can also use
countByKey method. The result returned by it is in hashmap (like dictionary) structure.
>>> package_count_2 = content.map(lambda x: (x, 1)).countByKey() >>> type(package_count_2) <type 'collections.defaultdict'> >>> package_count_2['ggplot2'] 3913 >>> package_count_2['stm'] 25
Please note that
countByKey method ONLY works on RDDs of type (K, V), returning a hashmap of (K, int) pairs with the COUNT of each key . The value of
V will NOT affect results from
countByKey! Just like the example below.
>>> package_count_2 = content.map(lambda x: (x, 1)).countByKey() >>> package_count_2['ggplot2'] 3913 >>> package_count_2_1 = content.map(lambda x: (x, 3)).countByKey() >>> package_count_2_1['ggplot2'] 3913 >>> package_count_2_2 = content.map(lambda x: (x, "test")).countByKey() >>> package_count_2_2['ggplot2'] 3913
After counting using
reduce method, I may want to know the rankings of these packages based on how many downloads they have. Then we need to use
sortByKey method. Please note:
- The 'Key' here refers to the first element of each tuple.
- The argument of
sortByKey(0 or 1) will determine if we're sorting descently ('0', or
False) or ascently ('1', or
# Sort DESCENTLY and get the first 10 >>> package_count.map(lambda x: (x, x)).sortByKey(0).take(10) [(4783, u'Rcpp'), (3913, u'ggplot2'), (3748, u'stringi'), (3449, u'stringr'), (3436, u'plyr'), (3265, u'magrittr'), (3223, u'digest'), (3205, u'reshape2'), (3046, u'RColorBrewer'), (3007, u'scales')] # Sort ascently and get the first 10 >>> package_count.map(lambda x: (x, x)).sortByKey(1).take(10) [(1, u'TSjson'), (1, u'ebayesthresh'), (1, u'parspatstat'), (1, u'gppois'), (1, u'JMLSD'), (1, u'kBestShortestPaths'), (1, u'StVAR'), (1, u'mosaicManip'), (1, u'em2'), (1, u'DART')]
Other than sorting by key (normally it's the first element in each tuple), we can also specify by which element to sort using method
>>> package_count.sortBy(lambda x:x).take(5) # default ascending is True [(u'TSjson', 1), (u'ebayesthresh', 1), (u'parspatstat', 1), (u'gppois', 1), (u'JMLSD', 1)] >>> package_count.sortBy(lambda x:x, ascending = False).take(5) [(u'Rcpp', 4783), (u'ggplot2', 3913), (u'stringi', 3748), (u'stringr', 3449), (u'plyr', 3436)]
We can consider
filter as the
SELECT * from TABLE WHERE ??? statement in SQL. It can help return a new dataset formed by selecting those elements on which the function specified by user returns
For example, I would like to obtain these downloading records of R package "Rtts" from China (CN), then the condition is "package == 'Rtts' AND country = 'CN'".
>>> content.filter(lambda x: x == 'Rtts' and x == 'CN').count() 1 >>> content.filter(lambda x: x == 'Rtts' and x == 'CN').take(1) [[u'2015-12-12', u'20:15:24', u'23820', u'3.2.2', u'x86_64', u'mingw32', u'Rtts', u'0.3.3', u'CN', u'41']]
Collect Result ('Export' into Python)
All the operations I listed above were done as RDD (Resilient Distributed Datasets). We can say that they were implemented 'within' Spark. And we may want to transfer some dataset into Python itself.
take method we used above can help us fulfill this purpose partially. But we also have
collect method to do this, and the difference between
take is that the former will return all the elements in the dataset by default and the later one will return the first
n rows (
n is specified by user). Meanwhile, we also need to be careful when we use
collect, since you may run out of your memory on Master node. In some references, it's suggested to NEVER use
collect() in production.
>>> temp = content.filter(lambda x: x == 'Rtts' and x == 'US').collect() >>> type(temp) <type 'list'> >>> temp [ [u'2015-12-12', u'04:52:36', u'23820', u'3.2.3', u'i386', u'mingw32', u'Rtts', u'0.3.3', u'US', u'1652'], [u'2015-12-12', u'20:31:45', u'23820', u'3.2.3', u'x86_64', u'linux-gnu', u'Rtts', u'0.3.3', u'US', u'4438'] ]
Like the set operators in vanilla SQL, we can do set operations in Spark. Here we would introduce
distinct. We can make intuitive interpretations as below.
- union of A and B: return elements of A AND elements of B.
- intersection of A and B: return these elements existing in both A and B.
- distinct of A: return the distinct values in A. That is, if element
aappears more than once, it will only appear once in the result returned.
>>> raw_content.count() 421970 # one set's union with itself equals to its "double" >>> raw_content.union(raw_content).count() 843940 # one set's intersection with itself equals to its disctinct value set >>> raw_content.intersection(raw_content).count() 421553 >>> raw_content.distinct().count() 421553
We may need to note that if each line of our data is an array instead of a string,
distinct methods can't work properly. This is why I used
raw_content instead of
content here as example.
Once again, I have found the data process methods in Spark is quite similar to that in SQL, like I can use
join method in Spark, which is a great news! Outer joins are also supported through
fullOuterJoin . Additionally,
cartesian is available as well (please note Spark SQL is available for similar purpose and would be preferred & recommended).
When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (V, W)) pairs with all pairs of elements for each key.
# generate a new RDD in which the 'country' variable is KEY >>> content_modified=content.map(lambda x:(x, x)) # give a mapping table of the abbreviates of four countries and their full name. >>> mapping=[('DE', 'Germany'), ('US', 'United States'), ('CN', 'China'), ('IN',"India")] >>> mapping=sc.parallelize(mapping) # join >>> content_modified.join(mapping).takeSample(False, 8) [ (u'CN', ([u'2015-12-12', u'19:26:01', u'512', u'NA', u'NA', u'NA', u'reweight', u'1.01', u'CN', u'4721'], 'China')), (u'US', ([u'2015-12-12', u'18:15:11', u'14271399', u'3.2.1', u'x86_64', u'mingw32', u'stringi', u'1.0-1', u'US', u'11837'], 'United States')), (u'US', ([u'2015-12-12', u'00:03:27', u'392370', u'3.2.3', u'x86_64', u'linux-gnu', u'colorspace', u'1.2-6', u'US', u'12607'], 'United States')), (u'US', ([u'2015-12-12', u'05:10:29', u'290932', u'3.2.2', u'x86_64', u'mingw32', u'iterators', u'1.0.8', u'US', u'5656'], 'United States')), (u'US', ([u'2015-12-12', u'22:28:47', u'2143454', u'3.2.3', u'x86_64', u'linux-gnu', u'quantreg', u'5.19', u'US', u'16318'], 'United States')), (u'US', ([u'2015-12-12', u'13:12:26', u'985806', u'3.2.3', u'x86_64', u'linux-gnu', u'plotly', u'2.0.3', u'US', u'2570'], 'United States')), (u'CN', ([u'2015-12-12', u'17:04:44', u'178399', u'3.2.1', u'x86_64', u'darwin13.4.0', u'apsrtable', u'0.8-8', u'CN', u'41'], 'China')), (u'US', ([u'2015-12-12', u'06:41:09', u'76007', u'3.2.3', u'i386', u'mingw32', u'superpc', u'1.09', u'US', u'1985'], 'United States')) ] # left outer join. # In the mapping table, we only gave the mappings of four countries, so we found some 'None' values in the returned result below >>> content_modified.leftOuterJoin(mapping).takeSample(False, 8) [ (u'US', ([u'2015-12-12', u'15:43:03', u'153892', u'3.2.2', u'i386', u'mingw32', u'gridBase', u'0.4-7', u'US', u'8922'], 'United States')), (u'CN', ([u'2015-12-12', u'19:59:37', u'82833', u'3.2.3', u'x86_64', u'mingw32', u'rgcvpack', u'0.1-4', u'CN', u'41'], 'China')), (u'JP', ([u'2015-12-12', u'17:24:59', u'2677787', u'3.2.3', u'i386', u'mingw32', u'ggplot2', u'1.0.1', u'JP', u'3597'], None)), (u'TN', ([u'2015-12-12', u'13:40:13', u'1229084', u'3.2.2', u'x86_64', u'mingw32', u'forecast', u'6.2', u'TN', u'10847'], None)), (u'US', ([u'2015-12-12', u'05:09:59', u'75327', u'3.2.3', u'x86_64', u'mingw32', u'xml2', u'0.1.2', u'US', u'5530'], 'United States')), (u'AE', ([u'2015-12-12', u'14:23:56', u'695625', u'3.1.2', u'i386', u'mingw32', u'mbbefd', u'0.7', u'AE', u'556'], None)), (u'KR', ([u'2015-12-12', u'16:31:34', u'36701', u'3.2.3', u'x86_64', u'linux-gnu', u'ttScreening', u'1.5', u'KR', u'4986'], None)), (u'US', ([u'2015-12-12', u'15:43:08', u'35212', u'3.2.2', u'x86_64', u'mingw32', u'reshape2', u'1.4.1', u'US', u'8922'], 'United States'))]
Some RDDs may be repeatedly accessed, like the RDD content in the example above. In such scenario, we may want to pull such RDDs into cluster-wide in-memory cache so that the computing relating to them will not be repeatedly invoked, so that resource and time can be saved. This is called "persisting" or "caching" in Spark, and can be done using
Spark automatically monitors cache usage on each node and drops out old data partitions in a least-recently-used (LRU) fashion. Of course we can also manually remove an RDD instead of waiting for it to fall out of the cache, using the RDD.unpersist() method.
We can also use
.is_cached to check whether a RDD is already cached or not.
>>> content.cache() >>> content.is_cached True >>> content.unpersist() >>> content.is_cached False #or >>> content.persist() >>> content.is_cached True >>> content.unpersist() >>> content.is_cached False
Please note caching may make little or even no difference when the data is small. In some case, persisting your RDDs may even make your application slower if the functions that computed your datasets are too simple. So do choose proper storage level when you persist your RDDs .
4. Submitting Application
All examples showed above were implemented interactively. To automate things, we may need to prepare scripts (applications) in advance and call them, instead of entering line by line.
spark-submit script in Spark’s
bin directory is just used to figure out this problem, i.e. launch applications. It can use all of Spark’s supported cluster managers (Scala, Java, Python, and R) through a uniform interface so you don’t have to configure your application specially for each one . This means that we only need to prepare and call the scripts while we don't need to tell Spark which driver we're using.
# submit application written with Python ./bin/spark-submit examples/src/main/python/pi.py # submit application written with R ./bin/spark-submit examples/src/main/r/dataframe.R
spark-submit, there are also several options we can specify, including which cluster to use (
--master) and arbitrary Spark configuration property. For details and examples of this, you may refere to Submitting Applications.
5. Spark SQL and DataFrames
Spark SQL is a Spark module for structured data processing . It enables users to run SQL queries on the data within Spark. DataFrame in Spark is conceptually equivalent to a table in a relational database or a data frame in R/Python . SQL queries in Spark will return results as DataFrames. Personal opinion, it's a bit more straightforward than RDD as DataFrame is just a TABLE itself.
For this section, please refer to a separate Jupyter Notebook file Spark DataFrames & SQL - Basics.
 Spark Programming Guide, http://spark.apache.org/docs/latest/programming-guide.html
 Submitting Applications, http://spark.apache.org/docs/latest/submitting-applications.html
 Spark Examples, http://spark.apache.org/examples.html
 Spark Configuration, http://spark.apache.org/docs/latest/configuration.html
 Spark SQL, DataFrames and Datasets Guide, http://spark.apache.org/docs/latest/sql-programming-guide.html
 Which Storage Level to Choose?, https://spark.apache.org/docs/latest/rdd-programming-guide.html#which-storage-level-to-choose
This work is licensed under the Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International License - CC BY-NC-SA 4.0