How to programe in pyspark on Pycharm locally, and execute the spark job remotely.

I have a Hadoop cluster of 4 worker nodes and 1 master node. Spark distribution (spark-1.4.0-bin-hadoop2.6) in installed on all nodes. I use ssh to connect to the master node from my laptop to execute Hadoop or Spark jobs on the cluster. The problem is that it is troublesome code through terminal, so I need an IDE that allows me to program on my laptop but execute the code on the cluster. This means that I need to access the remote master mode via this IDE. It is possible to use Eclipse for scala, details can be found here. As I use Python and PyCharm much often, in this blog, I describe how to enable this function in PyCharm. There is instructions on the website of PyCharm. But I hope to make things more detailed in this blog.

Steps we need to go through:

1. to enable PyCharm on my laptop to access to the the master node (or the remote server).

3. to add project interpreter

2. to access pyspark in PyCharm

Step 1. The thing is that we want to code locally with a nice interface, and to execute it remotely. This requires that the IDE is able to communicate with the remove server, that is the script files can be synchronized. By PyCharm, this configuration can be done in “Deployment” (Tools → Deployment → Configuration, or File → Settings → Build, Execution, Deployment → Deployment).

There are three tabs in the dialog window of “Deployment”: “Connection”, “Mappings” and “Executed Paths”. What we need to do is to specify the files in tabs “ Connection” and “Mappings”.

As shown in this screenshot, we need to specify the
(1) type of file transfer protocol, information for
(2) Upload/download project files and
(3) Browse files on server.

 First, in the name field, I put "master-minicluster" for the deployment. For the type of file transfer protocol, I chose SFTP (safe file transfer protocol) because my laptop and the master node both can communicate via password-less SSH. I filled the IP address of the master node in “SFTP host”. If you choose SFTP the Port is filled automatically with 22. For “Root path”, I gave the path of the user home on remote master node. “User name” I use on the master node is “hduser”. For security reason, you need to specify password or tell the path to the “private key file”. As for me, SSH has already establised before, I just need to give the path where the private key file id_rsa is saved. For “web server root URL”, just put http://IP_remote_server. If everything is right, you can press the button of “test SFTP connection”, a window should pop up saying that connection is successful.

 

Then for “Mappings” tab, we need to specify the paths of files that need to be synchronizes on the local machine and on the remote machine. I created the python project “test_Spark” locally, a folder with the same name is created on the master node. For the “web path on server”, I just add the IP address of the master node before the full path to “test_Spark” on it.

After the configuration of step 1, my laptop is able to access the master via PyCharm, and python files in the test_Spark folder should be synchronized between my laptop and the master node. But I can't run any program yet, because Python remote interpreter isn't added and pyspark path isn't specified.

Step2. Add project interpreter (ref: here). Python code that is written and debugged locally on my laptop is actually synchronized onto the master node and being executed on the cluster. This means the script is interpreted remotely, thus the path of the remote Python interpreter (for this project) should be specified in PyCharm.

Highlight the project “test_Spark”, and go to “File → Settings”, in the left column find “Project: test_Spark” and “Project Interpreter” under. On the right part, click the gear icon, a small window should pop up with three options: “add local”, “add remote” and “Create VirtualEnv”. Click “Add remote”, a dialog window should pop out shown as the figure below. 

 

In the dialog window, there are three ways to configure the remote python interpreter. As we did the configuration of Deployment, we can just click the third button and select “master-minicluster”. The “Python interpreter path” should be the one on the master node.

After step 2, program synchronized onto the master node should be executed in Python. Next, we need to specify SPARK_HOME and PYTHONPATH, so that we can use Spark and execute a job on the cluster.

Step 3. pyspark in PyCharm. There are two ways to do this, as shown here. The first way is that we specify the SPARK_HOME and append PYTHONPATH each time we call “pyspark”. That means we write the following lines at the beginning of the .py file.

 import os  
 import sys  
 # Path for spark source folder  
 os.environ['SPARK_HOME']="/home/hduser/spark-1.4.0-bin-hadoop2.6"  
 # Append pyspark to Python Path  
 sys.path.append("/home/hduser/spark-1.4.0-bin-hadoop2.6/python")  
 from pyspark import SparkContext  
 from pyspark import SparkConf  

Another way is to add the SPARK_HOME and PYTHONPATH in “Environment variables”. Highlight one .py file in the project folder, then go to “Run → Edit configuration”. In the tab of “Configuration”, find the “Environment”, and click the “...” in the field of “Environment variables”. A dialog window should pop up with only one line “PYTHONUNBUFFERED” by default, we need to click the “+” mark on top right and add SPARK_HOME and PYTHONPATH directing to the corresponding path in the master node, so don't forget “ssh://user_name@IP_remote_node/” before the path. SPARK_HOME is just the folder when you run and configure Spark. For Pyspark, you need go to the “python/lib” folder under SPARK_HOME, where you will find a “py4j-0.X.X.X-src.zip” file, you need to add “ py4j-0.X.X.X-src.zip$PYTHONPATH” in specifying the PYTHONPATH, shown as the screenshot, maybe your versions vary, change it accordingly.

 
After step 3, you should be able to import pyspark in your .py file.

Example
Let's run the word count example to see how everything works. I created a python file named as “WordCount.py” under “test_Spark” projet, shown as the screenshot below. In this file, I specified the SPARK_HOME and PYTHONPATH by adding some lines at the beginning (shown in step 3), so that I can import pyspark.

Next I need to upload WordCount.py onto the “test_Spark” folder on the remote master node. To do this, just right click “WordCount.py” in the project tree, and click “upload to master-minicluster”, then you will see file transfer information appears below. The upload was successful. I visited the “test_Spark” folder on the master node, this WordCount.py appeared. Then you just need to run “WordCount.py” in PyCharm, the job is executed on the remote cluster. You will see a progress bar appears under you code in PyCharm. In this example, we won't see the results in PyCharm, but we can use SSH to access the master and run “hadoop fs -cat” command to see the results via terminal.

If you modify WordCount.py, just ctrl+s, all changes will be synchronized with the WordCount.py on the remote machine.

Code for word count:

 import os  
 import sys  
 # Path for spark source folder  
 os.environ['SPARK_HOME']="/home/hduser/spark-1.4.0-bin-hadoop2.6"  
 # Append pyspark to Python Path  
 sys.path.append("/home/hduser/spark-1.4.0-bin-hadoop2.6/python")  
 from pyspark import SparkContext  
 from pyspark import SparkConf  
 sc = SparkContext(master="spark://IP_address_of_master_node:7077", appName="WordCount")  
 text_file = sc.textFile("hdfs://master:9000/user/xywang/textsForWordCount")  
 counts = text_file.flatMap(lambda line: line.split(" ")) \  
        .map(lambda word: (word, 1)) \  
        .reduceByKey(lambda a, b: a + b)  
 counts.saveAsTextFile("hdfs://master:9000/user/output/SparkWordCount/SparkResult6")  
 sc.stop()  

PS: To run this job, you need to have Hadoop and Spark turned on. And If you want to try the example, make sure to change your paths accordingly.

Troubleshooting: if error returns saying that "transfer failed... permission denied", you need to assign permission to the files in test_Spark folder on the remote server. For Ubuntu, simply type "$ sudo chmod 777 filename" in the terminal, thus you don't need to sudo when edit this file, do so to make the file synchronization work via ssh.

General steps of pre-processing in NLP and common problems met in dealing with French texts

We know that in NLP (Natural Language Processing), if we use the assumption of "bag-of-words" (the order of words doesn't matter), before we start any analysis (i.e. clustering, classification, topic extraction, etc.) on a collection of documents, let's name this collection as "corpus", a few things have to be done. General steps contains:

1. Lowercase all the text in each document.
2. Remove punctuation. 
3. Remove stop words, such as "a", "is", "that"... in English, or "un", "que", "suis"... in French. These words don't carry any real meanings but appear quite often.
4. Tokenization. It simply cut "cats are walking" into "cats" "are" "walking".
5. Stemming/Lemmatisation. In most cases you can do one of them, but not both. Stemming and lemmatisation are different. Stemming can convert "walking/walked" into "walk", but it can't do something lemmatisation can do, like converting "women" to "woman". 
6. So far, each document would be a set of words, if we define the pre-processed words are terms, we can actually assign an ID to each term.
7. Represent texts or say documents into Vector Space Model , so each text/document would be written into a form of vector. Inside the vector, there are term ID and a weight. This weight can be the number of occurrences of a term in a document, or it can be a if-idf weight. So "cats are walking" will be represented as [(0,1),(1,1)]. 0 is the ID of term "cat" (after stemming, "cats" becomes "cat"), the 1 after 0 says that term "cat" shows up once; "are" has been removed as a stop word. "walking" turns to "walk" after being stemmed, and "walk" has been assigned ID 1, as "walk" also appear only once, so its occurrence is also 1.

Very well. After the above 7 steps, you can basically turn a corpus into a big matrix, and with this matrix you can do whatever you want.

If you program with Python, you can have a few very good libraries, for example nltk, scikit-learn, gensim and so on. nltk is specialized in NLP with Python, it contains a lot of modules, which you can use to achieve any of the above steps with only a few lines of code; scikit-learn is a more general tool on Machine Learning (ML) with Python, it contains many ML algorithms, you may need a bit more to apply them onto texts; gensim is even more specialized, it is used to extract topic from a corpus. 

Dealing with English texts is well better than working on other languages. I have been working on French texts, French words that contains these letters é, ç, è, à, indeed gave me some hard time. Sometimes they just can't be shown properly in my SQL database, or they just can't be correctly encoded or decoded when I parsed an XML file. I found it might be tricky to use Notepad++, too. If you open your text with Notepad++, it is well easier to treat them if they are "Encode with UTF-8 without BOM". But most time they are originally in "Encode in ANSI" if your files are processed from Microsoft Word. 

Bigger problems occurs when it comes to stem French texts. If you are using Python nltk, it has Potter Stemming module, for English texts, you can just use this module and "Bingo", everything can be done. But it just can't be that easy with French texts, as French words have more different suffixes, and Potter Stemming doesn't support for French texts. Also it is more difficult to lemmatize French texts. 

Also in Part-of-Speech (POS), tagging French sentences becomes tricky too. And unfortunately, so far nltk doesn't have a well-implemented POS tag tool for French. Stanford NLP group has updated its POS tagger with an extension to French with Python, but I didn't find it very easy to use. After some exploration on Google, I luckily found TreeTagger, a language independent POS tagger. It is independent because it is using Markov Models. 

In my next blogger, I will explain how to use TreeTagger and POS tag French texts.