Update desc-pyspark to use Spark3

[heather999]

As discussed on Slack.
We have a new shifter image lsstdesc/desc-python:spark-v3.1.1 based on NERSC's most recent Spark docker image.
I would like to update the desc-pyspark jupyter kernel to use this version.
Additionally we will pull in Stephane's desc-spark scripts and install them along with the typical desc-python installations at NERSC.

[plaszczy]

I just tried the image on interactive nodes

cori08/~ $ salloc -N 5 -t 30 -C haswell -q debug --image=lsstdesc/desc-python:spark-v3.1.1 --volume='/global/cscratch1/sd/plaszczy/tmpfiles:/tmp:perNodeCache=size=200G'

but I get stuck on :

salloc: Granted job allocation 41422941
salloc: Waiting for resource configuration
salloc: Nodes nid0[1433-1437] are ready for job

in real bad way (even C^C nor C^z can help me!)

[heather999]

Thanks for giving this a try @plaszczy
I was able use the interactive queue rather than debug, and I used double quotes:

heatherk@cori12:/global/cscratch1/sd/heatherk> salloc -N 5 -t 30 -C haswell -q interactive --image=lsstdesc/desc-python:spark-v3.1.1 --volume="/global/cscratch1/sd/heatherk/tmpfiles:/tmp:perNodeCache=size=200G"
salloc: Granted job allocation 41429516
salloc: Waiting for resource configuration
salloc: Nodes nid000[09-13] are ready for job
heatherk@nid00009:/global/cscratch1/sd/heatherk> ls

Then I went back and tried the debug queue.. it took 5-10 minutes, but it finally started up:

heatherk@cori12:/global/cscratch1/sd/heatherk> salloc -N 5 -t 30 -C haswell -q debug --image=lsstdesc/desc-python:spark-v3.1.1 --volume="/global/cscratch1/sd/heatherk/tmpfiles:/tmp:perNodeCache=size=200G"
salloc: Pending job allocation 41429532
salloc: job 41429532 queued and waiting for resources
salloc: job 41429532 has been allocated resources
salloc: Granted job allocation 41429532
salloc: Waiting for resource configuration
salloc: Nodes nid0[0764,1262-1265] are ready for job
heatherk@nid00764:/global/cscratch1/sd/heatherk> 

Maybe try again and see if it works after a few minutes? The queue may have been busy.

[plaszczy]

yes you are right, I could log now.
But then I cannot have access to /root
ls /root/anaconda3/etc/profile.d/conda.sh
ls: cannot access '/root/anaconda3/etc/profile.d/conda.sh': No such file or directory

conda does not seems to be there
ls /root/
ansible.hash bin limits.conf limits.conf.2021-04-06@14:42:32

it does not seem to be the right python version
python -V
Python 2.7.17

[plaszczy]

actually it should be run within shifter. but
shifter "source /root/anaconda3/etc/profile.d/conda.sh"
shifter: source /root/anaconda3/etc/profile.d/conda.sh: No such file or directory

if we forget about the setup and just run (within the session)

shifter pyspark
env: ‘ipython’: Permission denied

so it seems that within the image , the user does not have access rights to the conda setup.

[heather999]

ah yes.. this is due to how the NERSC spark docker image is set up to install under /root. Let me go back to that NERSC ticket with Lisa and ask about that to get their suggestion. If I were setting up this image myself, I would have installed everything in another part of the directory tree to avoid that problem.

[heather999]

@plaszczy I see Lisa isn't available this week, so I went ahead and updated the docker image to reinstall Anaconda under /opt/desc/py to avoid the problems using /root. For now, this means we have two copies of Anaconda in the same image, but only ours in /opt/desc/py is made available by setting up PATH appropriately.
I've loaded this image lsstdesc/desc-python:spark-v3.1.1 into shifter and I can confirm it starts up and I can activate the conda environment by doing:

shifter -E --image=lsstdesc/desc-python:spark-v3.1.1 /bin/bash
$ source /opt/desc/py/etc/profile.d/conda.sh
$ conda activate base
(base) heatherk@cori03:/global/cfs/cdirs/lsst/gsharing/DC2_ImSim/DR2/repo/rerun/dr2-coadd$ 

Note to get pyspark to start up properly, I needed to include that -E when starting shifter to avoid loading my local environment.

Once Lisa is back and I get a new NERSC image we can clean up a bit, but I think this new lsstdesc/desc-python:spark-v3.1.1 might allow for some testing.

[plaszczy]

OK that's intresting, I could reproduce.

I avoid using -E that is very general, we found with Lisa that using:
export PYTHONUSERBASE=
avoids loading the users libs.

So after the source+activate conda, I ran a basic test in pyspark (which is an "ang2pix" in spark)

from pyspark.sql.functions import pandas_udf, PandasUDFType
import numpy as np
import pandas as pd
import healpy as hp

nside=256
nest=False
@pandas_udf('long', PandasUDFType.SCALAR)
def Ang2Pix(ra,dec):
    return pd.Series(hp.ang2pix(nside,np.radians(90-dec),np.radians(ra),nest=nest))

df = spark.createDataFrame([[10.0, 20]], ["ra", "dec"])
df=df.withColumn("ipix",Ang2Pix("ra","dec"))
df.show()

but we are back to the case pyspark does not find the libs:

...
  return self.loads(obj)
  File "/usr/local/bin/spark-3.1.1/python/lib/pyspark.zip/pyspark/serializers.py", line 430, in loads
    return pickle.loads(obj, encoding=encoding)
  File "/usr/local/bin/spark-3.1.1/python/lib/pyspark.zip/pyspark/cloudpickle/cloudpickle.py", line 562, in subimport
    __import__(name)
ModuleNotFoundError: No module named 'numpy'

[heather999]

HI @plaszczy
It's definitely something in the environment that carries over when we start shifter. PYTHONUSERBASE is already empty when we start the shifter image (whether or not we use the -E flag).
If I do this with -E, things appear to work:

heatherk@cori04:~> shifter -E --image=lsstdesc/desc-python:spark-v3.1.1 /bin/bash
<eatherk$ source /opt/desc/py/etc/profile.d/conda.sh               
heatherk@cori04:/global/u1/h/heatherk$ conda activate base
(base) heatherk@cori04:/global/u1/h/heatherk$ which java
/usr/bin/java
(base) heatherk@cori04:/global/u1/h/heatherk$ which pyspark
/usr/local/bin/spark-3.1.1/bin/pyspark
(base) heatherk@cori04:/global/u1/h/heatherk$ pyspark
Python 3.8.5 (default, Sep  4 2020, 07:30:14) 
Type 'copyright', 'credits' or 'license' for more information
IPython 7.19.0 -- An enhanced Interactive Python. Type '?' for help.
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/usr/local/bin/spark-3.1.1-bin-hadoop2.7/jars/spark-unsafe_2.12-3.1.1.jar) to constructor java.nio.DirectByteBuffer(long,int)
WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
21/04/08 06:49:45 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 3.1.1
      /_/

Using Python version 3.8.5 (default, Sep  4 2020 07:30:14)
Spark context Web UI available at http://cori04:4040
Spark context available as 'sc' (master = local[*], app id = local-1617889787587).
SparkSession available as 'spark'.

In [1]: from pyspark.sql.functions import pandas_udf, PandasUDFType

In [2]: import numpy as np

In [3]: import pandas as pd

In [4]: import healpy as hp

In [5]: nside=256
   ...: nest=False

In [6]: @pandas_udf('long', PandasUDFType.SCALAR)
   ...: def Ang2Pix(ra,dec):
   ...:     return pd.Series(hp.ang2pix(nside,np.radians(90-dec),np.radians(ra),
   ...: nest=nest))
   ...: 
/usr/local/bin/spark-3.1.1/python/pyspark/sql/pandas/functions.py:389: UserWarning: In Python 3.6+ and Spark 3.0+, it is preferred to specify type hints for pandas UDF instead of specifying pandas UDF type which will be deprecated in the future releases. See SPARK-28264 for more details.
  warnings.warn(

In [7]: df = spark.createDataFrame([[10.0, 20]], ["ra", "dec"])

In [8]: df=df.withColumn("ipix",Ang2Pix("ra","dec"))
   ...: df.show()
+----+---+------+                                                               
|  ra|dec|  ipix|
+----+---+------+
|10.0| 20|257564|
+----+---+------+

Let me see if I can figure out a way to get it to work without relying on that -E flag. Julien's set up for the pyspark jupyter kernel might offer me some hints: https://github.com/LSSTDESC/desc-python/blob/master/jupyter-kernels/desc-pyspark/desc-pyspark.sh