Jupyterhub with Multi Spark and Additional Kernals

from pyspark.sql import SparkSession

import random

# create / get spark session

spark = SparkSession.builder.appName("PiCalculator").getOrCreate()

sc = spark.sparkContext

# number of random points

NUM_SAMPLES = 1_000_000

def inside(_):

    x = random.random()

    y = random.random()

    return 1 if x*x + y*y <= 1 else 0

# parallelize and compute

count = sc.parallelize(range(NUM_SAMPLES)).map(inside).reduce(lambda a, b: a + b)

pi = 4.0 * count / NUM_SAMPLES

print("Pi is roughly:", pi)

print(spark.version)

import org.apache.spark.sql.SparkSession

import scala.util.Random

val spark = SparkSession.builder().getOrCreate()

val sc = spark.sparkContext

println("===== BASIC INFO =====")

println(s"Spark Version : ${sc.version}")

// Detect deploy mode

val isYarn = sc.master.startsWith("yarn") ||

             sc.getConf.get("spark.master", "").contains("yarn")

println(s"Running on YARN : $isYarn")

println("\n===== YARN PROOF (Selected Configs) =====")

val yarnKeys = Seq(

  "spark.master",

  "spark.submit.deployMode",

  "spark.yarn.app.id",

  "spark.yarn.queue"

)

yarnKeys.foreach { key =>

  val value = sc.getConf.getOption(key).getOrElse("NOT SET")

  println(s"$key = $value")

}

println("\n===== YARN ENV VARIABLES =====")

val yarnEnv = Seq(

  "HADOOP_CONF_DIR",

  "YARN_CONF_DIR",

  "SPARK_YARN_MODE",

  "CONTAINER_ID"

)

yarnEnv.foreach { key =>

  val value = sys.env.getOrElse(key, "NOT SET")

  println(s"$key = $value")

}

println("\n===== SPARK PI JOB =====")

val slices = 2

val n = math.min(100000L * slices, Int.MaxValue).toInt

val count = sc.parallelize(1 until n, slices).map { _ =>

  val x = Random.nextDouble()

  val y = Random.nextDouble()

  if (x*x + y*y <= 1) 1 else 0

}.reduce(_ + _)

val pi = 4.0 * count / (n - 1)

println(s"Pi estimate = $pi")

library(SparkR)

sparkR.session(master = "local[*]")

n <- 100000

partitions <- 2   # very small number of tasks

count <- spark.lapply(1:partitions, function(i) {

  inside <- 0

  for (j in 1:(n / partitions)) {

    x <- runif(1)

    y <- runif(1)

    if (x*x + y*y <= 1) inside <- inside + 1

  }

  inside

})

pi_estimate <- 4 * Reduce("+", count) / n

print(pi_estimate)

SELECT user();

create database doyle_hivetestnew;

create table doyle_hivetestnew.test(a int, b int);

show tables from doyle_hivetestnew;

insert into table doyle_hivetestnew.test(a, b) values (1, 1), (2, 4), (3, 9), (4, 16);

select * from doyle_hivetestnew.test;

import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

%reload_ext sql

from sqlalchemy import create_engine

engine = create_engine(

    "trino://kdoyle:passw0rd@newjupfin-0.newjupfin.harshith.svc.cluster.local:9098/hive",

    connect_args={

        "http_scheme": "https",

        "verify": False

    }

)

%sql engine

%%sql

show SCHEMAS

# ---- seperate cell ----#

%%sql

select * from hive.doyle_hivetestnew.test

# ---- seperate cell ----#

%%sql

INSERT INTO hive.doyle_hivetestnew.test (a, b) VALUES (5, 25)

# ---- seperate cell ----#

%%sql

select * from hive.doyle_hivetestnew.test