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()

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) _  

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

- 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 )

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)

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

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

ALTER TABLE table_identifier ADD [IF NOT EXISTS]
    ( partition_spec [ partition_spec ... ] )

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)

"""
A simple example demonstrating basic Spark SQL features using fictional
data inspired by a paper on determining the optimum length of chopsticks.
https://www.ncbi.nlm.nih.gov/pubmed/15676839
Run with:
  ./bin/spark-submit OptimumChopstick.py
"""
from __future__ import print_function
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.storagelevel import StorageLevel
#rdd.persist(StorageLevel.MEMORY_ONLY_SER)
​
# Get avg Food pinching effeciency by length
def AvgEffeciencyByLength(df):
    meansDf = df.groupby('ChopstickLength').mean('FoodPinchingEffeciency').orderBy('avg(FoodPinchingEffeciency)',ascending=0)
    return meansDf
​
# init
spark = SparkSession.builder.appName("Optimum Chopstick").getOrCreate()
sc = spark.sparkContext
input_loc = "s3://llubbe-gdelt-open-data/ChopstickEffeciency/"
​
# Read input by line
lines = sc.textFile(input_loc)
parts = lines.map(lambda l: l.split(","))
parts.persist(StorageLevel.MEMORY_ONLY_SER)
# Each line is converted to a tuple.
chopstickItems = parts.map(lambda p: (str(p[0]), float(p[1]), int(p[2]), int(p[3].strip())))
​
# Define a schema
fields = [StructField("TestID", StringType()),
          StructField("FoodPinchingEffeciency", DoubleType()), 
          StructField("Individual", IntegerType()), 
          StructField("ChopstickLength", IntegerType())]
schema = StructType(fields)
​
# Apply the schema to the RDD
chopsticksDF = spark.createDataFrame(chopstickItems, schema)
​
effeciencyByLength = AvgEffeciencyByLength(chopsticksDF)
effeciencyByLength.distinct().count()
​
moar_chopsticksDF = spark.read.load(input_loc, format="csv", schema=schema)
moar_effeciencyByLength = AvgEffeciencyByLength(moar_chopsticksDF)
moar_effeciencyByLength.distinct().count()
​
spark.stop()

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))

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('abc').getOrCreate()