<jupyter_start><jupyter_text>Reading and Processing Physics Data with Spark Example notebook to demonstrate the use of Jupyter notebooks, Spark-Root data source, Hadoop-XRootD connectorThe example in this notebook is mostly a repackaging of the notebook (https://github.com/vkhristenko/spark-root/blob/master/ipynb/publicCMSMuonia_exampleAnalysis.ipynb) by <NAME>   It uses CMS data from the CERN opendata portal (http://opendata.cern.ch)  Relevant techonology and links:  [spark-root](https://github.com/diana-hep/spark-root) - Data source for Spark to read data in [ROOT](https://root.cern.ch/) format   [Hadoop-XRootD connector](https://gitlab.cern.ch/awg/hadoop-xrootd-connector) - this is currently on CERN gitlab, will be made available also on GitHub  [CMS Big Data project](https://cms-big-data.github.io/)  [CERN openlab](http://openlab.cern/) data analytics project   Created: <EMAIL> August 2017  Prerequisites to set up the environment before starting this```export PATH=/opt/anaconda2/bin:$PATH this uses a local anaconda installtion for jupyter, python and pyspark installed with pip install note, normally we would use the envirnment on cvmfs, but currently it interferes with Hadoop-XRootD to be followed upsource /cvmfs/sft.cern.ch/lcg/etc/hadoop-confext/hadoop-setconf.sh  this sets up the YARN cluster variables, use the relevant cluster nameexport SPARK_DIST_CLASSPATH=/afs/cern.ch/user/c/canali/public/org.apache.bcel_bcel-6.0.jar      this is a temporary workaround till we fix cluster config to remove dependency to flumekinit check that KRB5CCNAME is set to FILE:Run the notebook and connect to it, for example by tunneling the port with something like:ssh -L 8888:localhost:8888 <EMAIL>jupyter notebook --no-browser```<jupyter_code># This creates the Spark Session with additional packages and environemnt variables
from pyspark.sql import SparkSession

spark=SparkSession.builder \
        .appName("CMS Big Data test notebook") \
        .config("spark.master","yarn") \
        .config("executor-memory","7g") \
        .config("spark.jars.packages", "org.diana-hep:spark-root_2.11:0.1.11,org.diana-hep:histogrammar-sparksql_2.11:1.0.3") \
        .config("spark.yarn.files", "$KRB5CCNAME#krbcache") \
        .config("spark.executorEnv.KRB5CCNAME","'FILE:$PWD/krbcache'") \
        .config("spark.driver.extraClassPath", "/usr/lib/hadoop/EOSfs.jar") \
        .config("spark.executor.extraClassPath", "/usr/lib/hadoop/EOSfs.jar") \
        .getOrCreate()

# The Spark session allocation above relies on the default config of the cluster which uses dynamic allocation 
# you can force allocating a fixed number of executors and cores using something like this example:
# .config("spark.dynamicAllocation.enabled","false").config("num-executors", 8).config("executor-cores", 4) \
# Test that Spark SQL works, debug code, can remove

sql = spark.sql
sql("select 'Hello World!'").show()
# Example 1: maps the root files stored in HDFS into a Spark Dataframe

df1 = spark.read \
        .format("org.dianahep.sparkroot") \
        .option("tree", "Events") \
        .load("hdfs:/cms/bigdatasci/vkhriste/data/publiccms_muionia_aod/0000/FEEFB039-0978-E011-BB60-E41F131815BC.root")
        
# Note, the example above reads only 1 file
# to read the entire folder, change it to: 
#   load("hdfs:/cms/bigdatasci/vkhriste/data/publiccms_muionia_aod/0000") to read the entire folder
# This is an actions, it reads the file and counts the events
df1.count()
df1.printSchema()
# Example 2: same as example 1 above, but this time is reads the root files stored in EOS into a Spark Dataframe
# This requires to have the Hadoop-XRootD connector installed on the driver and on the cluster

df2 = spark.read \
        .format("org.dianahep.sparkroot") \
        .option("tree", "Events") \
        .load("root://eospublic.cern.ch://eos/opendata/cms/Run2010B/MuOnia/AOD/Apr21ReReco-v1/0000/FEEFB039-0978-E011-BB60-E41F131815BC.root")
        
# Note, the example above reads only 1 file
# to read the entire folder, change it to: 
#   load("root://eospublic.cern.ch://eos/opendata/cms/Run2010B/MuOnia/AOD/Apr21ReReco-v1/0000/")

# This is an actions, it reads the file and counts the events
df2.count()
# Defines slimmedEvents Dataframe, a projection from the larger Dataframe df2
slimmedEvents = df2.select("recoMuons_muons__RECO_.recoMuons_muons__RECO_obj.reco::RecoCandidate.reco::LeafCandidate")
slimmedEvents.printSchema()
slimmedEvents.show(2,False)<jupyter_output>+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|reco::LeafCandidate                                                                                                                                                                                                                                                                                                                                                                                       |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------[...]<jupyter_text>Example of computation and visualization<jupyter_code>from math import *
import numpy
from pyspark.sql import Row
def invariantMass(muon1, muon2):
    pt1 = abs(muon1.pt_)
    phi1 = muon1.phi_
    theta1 = 2.0*atan(exp(-muon1.eta_))
    px1 = pt1 * cos(phi1)
    py1 = pt1 * sin(phi1)
    pz1 = pt1 / tan(theta1)
    E1 = sqrt(px1**2 + py1**2 + pz1**2 + 0.10565836727619171**2)
    #
    pt2 = abs(muon2.pt_)
    phi2 = muon2.phi_
    theta2 = 2.0*atan(exp(-muon2.eta_))
    px2 = pt2 * cos(phi2)
    py2 = pt2 * sin(phi2)
    pz2 = pt2 / tan(theta2)
    E2 = sqrt(px2**2 + py2**2 + pz2**2 + 0.10565836727619171**2)
    #
    return sqrt((E1 + E2)**2 - (px1 + px2)**2 - (py1 + py2)**2 - (pz1 + pz2)**2)

def handleEvent(event):
    # decreasing order of muon pT
    sortedMuons = sorted(event[0], key=lambda muon: -muon.pt_)
    if len(sortedMuons) < 2:
        return []
    else:
        muon1, muon2 = sortedMuons[:2]
        # return [Row(mass=invariantMass(muon1, muon2), pt1=muon1.pt_, phi1=muon1.phi_, eta1=muon1.eta_, pt2=muon2.pt_, phi2=muon2.phi_, eta2=muon2.eta_)]
        return [invariantMass(muon1, muon2)]
dimuon_masses = slimmedEvents.rdd.flatMap(handleEvent).map(lambda x: (x, )).toDF().persist()
# This is an action and can take a few seconds/minutes depending the data size and number of executors
dimuon_masses.count()
dimuon_masses.show(5)
# Setup matpolotlib for plotting
%pylab inline
pylab.rcParams['figure.figsize'] = (12.0, 8.0)
tmpdf = dimuon_masses.toDF("Invariant Mass").limit(1000000)

tmpdf.toPandas().plot.hist(bins=200, range=[0, 10])
xlabel('Bin value')
ylabel('Count');<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.19004905395935529, 'fraction_numerical': 0.03601962158374212, 'mean_word_length': 2.280919540229885, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 11, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '639559', 'license': 'Apache-2.0', 'n_tokens_mistral': 2332, 'n_tokens_neox': 2146, 'n_words': 554, 'path': 'Spark_Physics/misc/CMS_BigData_Opendata_Spark_Example1.ipynb', 'repo_name': 'Code360In/Miscellaneous', 'size': '7179'}