Document use of conda environments with GeoMesa, PySpark & Jupyter

Document use of conda environments with GeoMesa, PySpark & Jupyter

import findspark # Load Spark library from locally installed HDP version findspark.init("/usr/hdp/current/spark2-client/") import pyspark # Spark JVM config will send logs to stderr, but in Jupyter this will end up in the system journal which isn't very useful # since sys.stdout and sys.stderr are actually ipykernel.iostream.OutStreams :-( # This (modified) code comes from https://github.com/vericast/spylon-kernel/blob/master/spylon_kernel/scala_interpreter.py (BSD-3) # It is probably pyspark version specific, and might be better/different in Spark 3 # Annoyingly can't use Jupyter's mangled sys.stdout and sys.stderr directly since they don't look enough like files # Patch the pyspark.java_gateway.Popen instance to force it to pipe output # so that we can catch all output from the Spark driver JVM spark_jvm_proc = None import pyspark.java_gateway, subprocess def my_Popen(*args, **kwargs): """Wraps subprocess.Popen to force stdout and stderr from the child process to sys.stdout/err without buffering. """ print("Redirecting Spark driver output to pipe") global spark_jvm_proc # Override these in kwargs to avoid duplicate value errors # Set streams to unbuffered so that we read whatever bytes are available # when ready, https://docs.python.org/3.6/library/subprocess.html#popen-constructor kwargs['bufsize'] = 0 # Capture everything from stderr for display in the notebook via pipe kwargs['stderr'] = subprocess.PIPE spark_jvm_proc = subprocess.Popen(*args, **kwargs) return spark_jvm_proc pyspark.java_gateway.Popen = my_Popen # Create and start thread to poll stderr (default destination for log4j). This seems to put the logs in the right cell (!) import threading import time def read_spark_stderr(): while True: if spark_jvm_proc is not None: print(spark_jvm_proc.stderr.readline().decode('UTF-8')) time.sleep(0.05) spark_stderr_thread = threading.Thread(target = read_spark_stderr) spark_stderr_thread.start() import os from pyspark.sql import SparkSession # findspark sets PYSPARK_PYTHON to the driver python path (wrong) & spark.pyspark.python doesn't seem to work, so reset PYSPARK_PYTHON os.environ["PYSPARK_PYTHON"] = "./environment/bin/python" spark = ( SparkSession.builder .appName("GeoMesa PySpark") .config("spark.yarn.dist.archives", "/path/to/geomesa-jupyter.tar.gz#environment") .config("spark.yarn.dist.jars", "/path/to/geomesa-accumulo-spark-runtime_2.11-2.4.2.jar") .config("spark.driver.extraClassPath", "/path/to/geomesa-accumulo-spark-runtime_2.11-2.4.2.jar") .config("spark.dynamicAllocation.enabled", "true") .config("spark.shuffle.service.enabled", "true") .config("spark.dynamicAllocation.minExecutors", "10") .config("spark.dynamicAllocation.maxExecutors", "100") .config("spark.dynamicAllocation.cachedExecutorIdleTimeout", "1h") .config("spark.executor.instances", "10") .config("spark.executor.memory", "8g") .getOrCreate() ) spark # Includes handy link to Spark UI

Original JIRA Issue: https://geomesa.atlassian.net/browse/GEOMESA-3097

Key: GEOMESA-3097
Type: Improvement
Priority: Major
Status: To Do
Resolution: Unresolved
Reporter: James Srinivasan
Created: Tue, 6 Jul 2021 16:14:20 -0400
Updated: Thu, 8 Jul 2021 10:48:45 -0400

Attachments:

• GeoMesa & Jupyter-v3-20210706_161252.pdf (54.9 KB) - uploaded by jsrinivasan on Tue, 6 Jul 2021 16:14:45 -0400

Start by locating the GeoMesa documentation area for Jupyter, PySpark, or environment setup; no repository file or test is named in the issue. Use the supplied conda, findspark, SparkSession, and YARN archive example as the source material, and consider the work done when a newcomer can follow the documented setup and run GeoMesa with PySpark in Jupyter.