Using Spark 2 from Python

# Word count
# 
# This example shows how to count the occurrences of each word in a text file.
  
from __future__ import print_function
import sys, re
from operator import add
from pyspark.sql import SparkSession
  
spark = SparkSession\
  .builder\
  .appName("PythonWordCount")\
  .getOrCreate()
  
# The file you use as input must already exist in HDFS.
# Add the data file to hdfs.
!hdfs dfs -put resources/cgroup-v2.txt /tmp

# Access the file from wordcount.py  
lines = spark.read.text("/tmp/cgroup-v2.txt").rdd.map(lambda r: r[0])
counts = lines.flatMap(lambda x: x.split(' ')) \
  .map(lambda x: (x, 1)) \
  .reduceByKey(add) \
  .sortBy(lambda x: x[1], False)
output = counts.collect()
for (word, count) in output:
  print("%s: %i" % (word, count))

spark.stop()
 

1. (Prerequisites)
   • Make sure the Anaconda parcel has been distributed and activated on your cluster.
   • Create a new project in Cloudera Data Science Workbench. In that project, create a new file called nltk_sample.py with the following sample script.
     nltk_sample.py

     # This code uses NLTK, a Python natural language processing library.
     # NLTK is not installed with conda by default. 
     # You can use 'import' within your Python UDFs, to use Python libraries.
     # The goal here is to distribute NLTK with the conda environment.
     
     import os
     import sys
     from pyspark.sql import SparkSession
      
     spark = SparkSession.builder \
               .appName("spark-nltk") \
               .getOrCreate()
      
     data = spark.sparkContext.textFile('1970-Nixon.txt')
      
     def word_tokenize(x):
         import nltk
         return nltk.word_tokenize(x)
      
     def pos_tag(x):
         import nltk
         return nltk.pos_tag([x])
      
     words = data.flatMap(word_tokenize)
     words.saveAsTextFile('nixon_tokens')
      
     pos_word = words.map(pos_tag)
     pos_word.saveAsTextFile('nixon_token_pos')

2. Go to the project you created and launch a new PySpark session.
3. Click Terminal Access and run the following command to pack the Python environment into conda.

   conda create -n nltk_env --copy -y -q python=2.7.11 nltk numpy

   The --copy option allows you to copy whole dependent packages into certain directory of a conda environment. If you want to add extra pip packages without conda, you should copy packages manually after using pip install. In Cloudera Data Science Workbench, pip will install packages into the ~/.local directory in a project.

   pip install some-awesome-package
   cp -r ~/.local/lib ~/.conda/envs/nltk_env/

   Zip the conda environment for shipping on PySpark cluster.

   cd ~/.conda/envs
   zip -r ../../nltk_env.zip nltk_env

4. (Specific to NLTK) For this example, you can use NLTK data as input.

   cd ~/
   source activate nltk_env
   
   # download nltk data
   (nltk_env)$ python -m nltk.downloader -d nltk_data all
   (nltk_env)$ hdfs dfs -put nltk_data/corpora/state_union/1970-Nixon.txt ./
     
   # archive nltk data for distribution
   cd ~/nltk_data/tokenizers/
   zip -r ../../tokenizers.zip *
   cd ~/nltk_data/taggers/
   zip -r ../../taggers.zip *

5. Set spark-submit options in spark-defaults.conf.

   spark.yarn.appMasterEnv.PYSPARK_PYTHON=./NLTK/nltk_env/bin/python
   spark.yarn.appMasterEnv.NLTK_DATA=./
   spark.executorEnv.NLTK_DATA=./
   spark.yarn.dist.archives=nltk_env.zip#NLTK,tokenizers.zip#tokenizers,taggers.zip#taggers

   With these settings, PySpark unzips nltk_env.zip into the NLTK directory. NLTK_DATA is the environmental variable where NLTK data is stored.

6. In Cloudera Data Science Workbench, set the PYSPARK_PYTHON environment variable to the newly-created environment. To do this, navigate back to the Project Overview page and click Settings > Engine > Environment Variables. Set PYSPARK_PYTHON to ./NLTK/nltk_env/bin/python and click Add. Then click Save Environment.
7. Launch a new PySpark session and run the nltk_sample.py script in the workbench. You can test whether the script ran successfully using the following command:

   !hdfs dfs -cat ./nixon_tokens/* | head -n 20
   Annual
   Message
   to
   the
   Congress
   on
   the
   State
   of
   the
   Union
   .
   January
   22
   ,
   1970
   Mr.
   Speaker
   ,
   Mr.
     
   ! hdfs dfs -cat nixon_token_pos/* | head -n 20
   [(u'Annual', 'JJ')]
   [(u'Message', 'NN')]
   [(u'to', 'TO')]
   [(u'the', 'DT')]
   [(u'Congress', 'NNP')]
   [(u'on', 'IN')]
   [(u'the', 'DT')]
   [(u'State', 'NNP')]
   [(u'of', 'IN')]
   [(u'the', 'DT')]
   [(u'Union', 'NN')]
   [(u'.', '.')]
   [(u'January', 'NNP')]
   [(u'22', 'CD')]
   [(u',', ',')]
   [(u'1970', 'CD')]
   [(u'Mr.', 'NNP')]
   [(u'Speaker', 'NN')]
   [(u',', ',')]
   [(u'Mr.', 'NNP')]