How can I connecta postgreSQL database into Apache Spark using scala? - Stack Overflow

import org.apache.spark.sql.functions._

val sc: SparkContext = ...
val sqlContext = new SQLContext(sc)

import sqlContext.implicits._

val input = sc.parallelize(Seq(
  ("a", 5, 7, 9, 12, 13),
  ("b", 6, 4, 3, 20, 17),
  ("c", 4, 9, 4, 6 , 9),
  ("d", 1, 2, 6, 8 , 1)
)).toDF("ID", "var1", "var2", "var3", "var4", "var5")

val columnsToSum = List(col("var1"), col("var2"), col("var3"), col("var4"), col("var5"))

val output = input.withColumn("sums", columnsToSum.reduce(_ + _))

output.show()

split_col = pyspark.sql.functions.split(df['my_str_col'], '-')
df = df.withColumn('NAME1', split_col.getItem(0))
df = df.withColumn('NAME2', split_col.getItem(1))

val classes = Seq(
  getClass,                   // To get the jar with our own code.
  classOf[mysql.jdbc.Driver]  // To get the connector.
)
val jars = classes.map(_.getProtectionDomain().getCodeSource().getLocation().getPath())
val conf = new SparkConf().setJars(jars)

postgres-# ( terminal commands)
- list all databases : \l
- list of all tables : \d
- list only the tables : \dt
- list all fields in a table : \d person (name of the table)
- import from external file : \i c:/postgrescourse/person.sql

- connect to database psql and import .sql file
> psql -h localhost -U postgres -d forgeprod -f C:\forgeprod.sql
https://kb.objectrocket.com/postgresql/how-to-run-an-sql-file-in-postgres-846

- access to database to query 
> psql -U postgres -d forgeprod

- datatypes info:
bigserial : increment int in sequence
date : year,month,day



- create database : create database test;
- drop database : drop database test;
- drop table : drop table person;
- connect to a database ( two forms ) :  
\c test (or) -h localhost -p 5432 -U postgres test
- create table sintax : create table table_name ( column name + data type + constraints if any ) 

- create table : 
create table person ( 
	id int,
	first_name varchar(50),
  	last_name varchar(50),
    gender varchar(6),
    date_of_birth date
    );
- create table w/constraints ( need to satisfied rules): 
create table person (
 	id bigserial not null primary key,
 	first_name varchar(50) not null,
 	last_name varchar(50) not null,
    gender varchar(6) not null, 
    date_of_birth date not null
	);

- create table : 
insert into person ( first_name, last_name, gender, date_of_birth) values ('Anne','Smith','Female',date '1988-01-09');

- Using OFFSET and LIMIT : select * from person OFFSET 5 LIMIT 5;
- Using OFFSET and FETCH : select * from person OFFSET 5 FETCH 5 ROW ONLY;
- Using BETWEEN : select * from person where date_of_birth BETWEEN DATE '2000-01-01' and '2015-01-01';
- Diferences between LIKE and ILIKE : ILIKE is case insensitive. 

- Using Group By : select country_of_birth, count(*) from person group by country_of_birth; ( will count how many people are from that country )
- Using Having with Group By : select country_of_birth, count(*) from person having count(*) > 5 group by country_of_birth; ( must have above 5 to show )

/* SQL Command within postgres: */

SELECT now() - pg_postmaster_start_time() uptime

/* using psql command line utility deployed with Postgres: */
 
$> psql -c "SELECT now() - pg_postmaster_start_time() uptime"

# remove references to the deprecated value
UPDATE job SET job_status = 'running' WHERE job_status = 'waiting';

# rename the existing type
ALTER TYPE status_enum RENAME TO status_enum_old;

# create the new type
CREATE TYPE status_enum AS ENUM('queued', 'running', 'done');

# update the columns to use the new type
ALTER TABLE job ALTER COLUMN job_status TYPE status_enum USING job_status::text::status_enum;
# if you get an error, see bottom of post

# remove the old type
DROP TYPE status_enum_old;

val keysDF = mapTypeDF.select(explode(map_keys($"property"))).distinct()
val keys = keysDF.collect().map(f=>f.get(0))
val keyCols = keys.map(f=> col("property").getItem(f).as(f.toString))
val expandedDf = mapTypeDF.select(col("name") +: keyCols:_*).show(false)

val columnsToSum = List(col("var1"), col("var2"), col("var3"), col("var4"), col("var5"))

val output = input.withColumn("sums", columnsToSum.reduce(_ + _))

def mkPropEqMatcher[T,V](name:String, extr: T => V)(expVal : V): HavePropertyMatcher[T,V] =
     new HavePropertyMatcher[T, V] {
       def apply(t: T): HavePropertyMatchResult[V] =
         HavePropertyMatchResult(
           extr(t) == expVal,
           name,
           expVal,
           extr(t)
         )
     }
 
   val title = mkPropEqMatcher("title", (_ : Book).title) _
   val author = mkPropEqMatcher("author", (_ : Book).author) _  

n match {
  case it if 0 until 5 contains it  => "less than five"
  case it if 5 until 10 contains it => "less than ten"
  case _ => "a lot"
}

class Contains(r: Range) { def unapply(i: Int): Boolean = r contains i }

val C1 = new Contains(3 to 10)
val C2 = new Contains(20 to 30)

scala> 5 match { case C1() => println("C1"); case C2() => println("C2"); case _ => println("none") }
C1

scala> 23 match { case C1() => println("C1"); case C2() => println("C2"); case _ => println("none") }
C2

scala> 45 match { case C1() => println("C1"); case C2() => println("C2"); case _ => println("none") }
none

import org.openjdk.jmh.annotations.{Benchmark, BenchmarkMode, Fork, Measurement, Mode, Param, Scope, State, Warmup}

import java.util.ArrayDeque
import scala.util.Random
import org.openjdk.jmh.annotations.Setup
import scala.jdk.StreamConverters.*

case class Cls(i: Int)

@BenchmarkMode(Array(Mode.Throughput))
@State(Scope.Benchmark)
@Measurement(time = 1, iterations = 5)
@Warmup(time = 1, iterations = 5)
@Fork(value = 2)
class Bench:

  @Param(Array("100", "10000", "100000", "1000000"))
  var size: Int = _

  var data: Array[Cls] = _
  var list: List[Cls] = _
  var vector: Vector[Cls] = _

  @Setup
  def setup(): Unit =
    data = Array.fill(size)(Cls(Random.nextInt))
    list = List.from(data)
    vector = Vector.from(data)

  @Benchmark
  def listBuilderAndSum: Long =
    val builder = List.newBuilder[Cls]
    for cls <- data do builder.addOne(cls)
    val list = builder.result()

    list.filter(_.i > 0).map(_.i.toLong).sum


  @Benchmark
  def vectorBuilderAndSum: Long =
    val builder = Vector.newBuilder[Cls]
    for i <- data do builder.addOne(i)
    val vec = builder.result

    vec.filter(_.i > 0).map(_.i.toLong).sum

  @Benchmark
  def vectorBuildDirectAndSum: Long =
    var vec = Vector.empty[Cls]
    for cls <- data do vec = vec :+ cls

    vec.filter(_.i > 0).map(_.i.toLong).sum


  @Benchmark
  def listPrependAndSum: Long =
    var lst = List.empty[Cls]
    for cls <- data do lst = cls :: lst
    lst.filter(_.i > 0).map(_.i.toLong).sum

  @Benchmark
  def vectorPrependAndSum: Long =
    var vec = Vector.empty[Cls]
    for cls <- data do vec = cls +: vec
    vec.filter(_.i > 0).map(_.i.toLong).sum
  
  @Benchmark
  def listSum: Long =
    list.filter(_.i > 0).map(_.i.toLong).sum

  @Benchmark
  def vectorSum: Long =
    vector.filter(_.i > 0).map(_.i.toLong).sum

  @Benchmark
  def listViewSum: Long =
    list.view.filter(_.i > 0).map(_.i.toLong).sum

  @Benchmark
  def vectorViewSum: Long =
    vector.view.filter(_.i > 0).map(_.i.toLong).sum

  @Benchmark
  def listJavaStreamSum: Long =
    list.asJavaSeqStream.filter(_.i > 0).mapToLong(_.i.toLong).sum

  @Benchmark
  def vectorJavaStreamSum: Long =
    vector.asJavaSeqStream.filter(_.i > 0).mapToLong(_.i.toLong).sum

import scala.compiletime.{error, requireConst}

opaque type NonEmptyString = String

object NonEmptyString:
  def apply(s: String): Option[NonEmptyString] = 
    if s.isEmpty then None else Some(s)

  inline def from(inline s: String): NonEmptyString =
    requireConst(s)
    inline if s == "" then error("got an empty string") else s

  given Conversion[NonEmptyString, String] with
    inline def apply(nes: NonEmptyString): String = nes  

opaque type LowerCased = String

object LowerCased:
  def apply(s: String): LowerCased = s.toLowerCase(Locale.US)

  given Conversion[LowerCased, String] with
    inline def apply(lc: LowerCased): String = lc

object LowerCased:
  inline def from(inline s: String): LowerCased = ${ fromImpl('s) }

  import scala.quoted.*
  private def fromImpl(s: Expr[String])(using Quotes): Expr[LowerCased] = 
    ???