spark_1.x_sql_examples_2

----------------------------------------------------------------------------------
SPARK SQL EXAMPLES with SCALA / JAVA / PYTHON / R
----------------------------------------------------------------------------------


Usage of "SqlContext" in scala / java
----------------------------------------------------------------------------------

Scala:
-----------
val sc: SparkContext = // An existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// Create the DataFrame
val df = sqlContext.read.json("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json")


Java:
-----------
JavaSparkContext sc = ...; // An existing JavaSparkContext.
SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);

// Create the DataFrame
DataFrame df = sqlContext.read().json("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json");



// Show the content of the DataFrame
df.show()

+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+


// Print the schema in a tree format
df.printSchema()

root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)



// Select only the "name" column
df.select("name").show()

+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+


// Select everybody, but increment the age by 1
df.select(df("name"), df("age") + 1).show()

+-------+---------+
|   name|(age + 1)|
+-------+---------+
|Michael|     null|
|   Andy|       31|
| Justin|       20|
+-------+---------+


// Select people older than 21
df.filter(df("age") > 21).show()

+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+



// Count people by age
df.groupBy("age").count().show()

+----+-----+
| age|count|
+----+-----+
|null|    1|
|  19|    1|
|  30|    1|
+----+-----+



Usage of "SqlContext" in python
=========================================
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

# Create the DataFrame
df = sqlContext.read.json("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json")


# Show the content of the DataFrame
df.show()

+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+



# Print the schema in a tree format
df.printSchema()

root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)



# Select only the "name" column
df.select("name").show()

+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+



# Select everybody, but increment the age by 1
df.select(df['name'], df['age'] + 1).show()

+-------+---------+
|   name|(age + 1)|
+-------+---------+
|Michael|     null|
|   Andy|       31|
| Justin|       20|
+-------+---------+



# Select people older than 21
df.filter(df['age'] > 21).show()

+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+


# Count people by age
df.groupBy("age").count().show()

+----+-----+
| age|count|
+----+-----+
|null|    1|
|  19|    1|
|  30|    1|
+----+-----+



Usage of "SqlContext" in R
=========================================
sqlContext <- sparkRSQL.init(sc)

# Create the DataFrame
df <- jsonFile(sqlContext, "file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json")

df <- read.json(sqlContext, "file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json")


# Show the content of the DataFrame
showDF(df)

+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+


# Print the schema in a tree format
printSchema(df)

root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)


# Select only the "name" column
showDF(select(df, "name"))

+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+


# Select everybody, but increment the age by 1
showDF(select(df, df$name, df$age + 1))

+-------+-----------+
|   name|(age + 1.0)|
+-------+-----------+
|Michael|       null|
|   Andy|       31.0|
| Justin|       20.0|
+-------+-----------+


# Select people older than 21
showDF(where(df, df$age > 21))

+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+


# Count people by age
showDF(count(groupBy(df, "age")))

+----+-----+
| age|count|
+----+-----+
|null|    1|
|  19|    1|
|  30|    1|
+----+-----+




Creating Datasets using scala:
=========================================
// Encoders for most common types are automatically provided by importing sqlContext.implicits._

val ds = Seq(1, 2, 3).toDS()

ds.map(_ + 1).collect()
res: Array[Int] = Array(2, 3, 4)


// Encoders are also created for case classes.

case class Person(name: String, age: Long)

val ds = Seq(Person("Andy", 32)).toDS()


// DataFrames can be converted to a Dataset by providing a class.
// Mapping will be done by name.

val path = "file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json"

val people1 = sqlContext.read.json(path)
people1.collect()
res: Array[org.apache.spark.sql.Row] = Array([null,Michael], [30,Andy], [19,Justin])


val people2 = sqlContext.read.json(path).as[Person]
people2.collect()
res: Array[Person] = Array(Person(Michael,-1), Person(Andy,30), Person(Justin,19))


Inferring the Schema Using Reflection (using scala code)
===========================================================
// Create an RDD of Person objects and register it as a table.
val people = sc.textFile("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()

people.registerTempTable("people")


// SQL statements can be run by using the sql methods provided by sqlContext.
val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")

// The results of SQL queries are DataFrames and support all the normal RDD operations.
// The columns of a row in the result can be accessed by field index:
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)
res: Name: Justin


// or by field name:
teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)
res: Name: Justin


// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)
res: Map("name" -> "Justin", "age" -> 19)



Inferring the Schema Using Reflection (using python code)
===========================================================
# sc is an existing SparkContext.
from pyspark.sql import SQLContext, Row
sqlContext = SQLContext(sc)


# Load a text file and convert each line to a Row.
lines = sc.textFile("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))


# Infer the schema, and register the DataFrame as a table.
schemaPeople = sqlContext.createDataFrame(people)
schemaPeople.registerTempTable("people")


# SQL can be run over DataFrames that have been registered as a table.
teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")


# The results of SQL queries are RDDs and support all the normal RDD operations.
teenNames = teenagers.map(lambda p: "Name: " + p.name)
for teenName in teenNames.collect():
  print(teenName)


Programmatically Specifying the Schema (using scala code)
==============================================================

// Create an RDD
val people = sc.textFile("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.txt")


// The schema is encoded in a string
val schemaString = "name age"


// Import Row.
import org.apache.spark.sql.Row;


// Import Spark SQL data types
import org.apache.spark.sql.types.{StructType,StructField,StringType};


// Generate the schema based on the string of schema
val schema = StructType( schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))


// Convert records of the RDD (people) to Rows.
val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))

// Apply the schema to the RDD.
val peopleDataFrame = sqlContext.createDataFrame(rowRDD, schema)


// Register the DataFrames as a table.
peopleDataFrame.registerTempTable("people")


// SQL statements can be run by using the sql methods provided by sqlContext.
val results = sqlContext.sql("SELECT name FROM people")


// The results of SQL queries are DataFrames and support all the normal RDD operations.
// The columns of a row in the result can be accessed by field index or by field name.
results.map(t => "Name: " + t(0)).collect().foreach(println)




Programmatically Specifying the Schema (using python code)
==============================================================

# Import SQLContext and data types
from pyspark.sql import SQLContext
from pyspark.sql.types import *

# sc is an existing SparkContext.
sqlContext = SQLContext(sc)


# Load a text file and convert each line to a tuple.
lines = sc.textFile("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: (p[0], p[1].strip()))


# The schema is encoded in a string.
schemaString = "name age"


fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
schema = StructType(fields)


# Apply the schema to the RDD.
schemaPeople = sqlContext.createDataFrame(people, schema)


# Register the DataFrame as a table.
schemaPeople.registerTempTable("people")


# SQL can be run over DataFrames that have been registered as a table.
results = sqlContext.sql("SELECT name FROM people")


# The results of SQL queries are RDDs and support all the normal RDD operations.
names = results.map(lambda p: "Name: " + p.name)
for name in names.collect():
  print(name)


Generic Load/Save Functions
========================================
scala:
--------------
val df = sqlContext.read.load("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/users.parquet")
df.select("name", "favorite_color").write.save("file:///home/orienit/spark/output/namesAndFavColors.parquet")


python:
--------------
df = sqlContext.read.load("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/users.parquet")
df.select("name", "favorite_color").write.save("file:///home/orienit/spark/output/namesAndFavColors.parquet")


R:
--------------
df <- loadDF(sqlContext, "file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.parquet")
saveDF(select(df, "name", "age"), "file:///home/orienit/spark/output/namesAndAges.parquet")





Manually Specifying Options
=============================================
scala:
--------------
val df = sqlContext.read.format("json").load("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json")
df.select("name", "age").write.format("parquet").save("file:///home/orienit/spark/output/namesAndAges.parquet")
df.select("name", "age").write.format("json").save("file:///home/orienit/spark/output/namesAndAges.json")


python:
--------------
df = sqlContext.read.load("file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json", format="json")
df.select("name", "age").write.save("file:///home/orienit/spark/output/namesAndAges.parquet", format="parquet")


R:
--------------
df <- loadDF(sqlContext, "file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/people.json", "json")
saveDF(select(df, "name", "age"), "file:///home/orienit/spark/output/namesAndAges.parquet", "parquet")




Run SQL on files directly
=============================================
scala:
--------------
val df = sqlContext.sql("SELECT * FROM parquet.`file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/users.parquet`")


python:
--------------
df = sqlContext.sql("SELECT * FROM parquet.`file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/users.parquet`")


R:
--------------
df <- sql(sqlContext, "SELECT * FROM parquet.`file:///home/orienit/spark/spark-1.6.0-bin-hadoop2.6/examples/src/main/resources/users.parquet`")



Loading Data Programmatically
=============================================
scala:
--------------
val people: RDD[Person] = ... // An RDD of case class objects, from the previous example.


// The RDD is implicitly converted to a DataFrame by implicits, allowing it to be stored using Parquet.
people.write.parquet("people.parquet")


// Read in the parquet file created above. Parquet files are self-describing so the schema is preserved.
// The result of loading a Parquet file is also a DataFrame.
val parquetFile = sqlContext.read.parquet("people.parquet")


//Parquet files can also be registered as tables and then used in SQL statements.
parquetFile.registerTempTable("parquetFile")
val teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)



python:
--------------
# sqlContext from the previous example is used in this example.

schemaPeople # The DataFrame from the previous example.


# DataFrames can be saved as Parquet files, maintaining the schema information.
schemaPeople.write.parquet("people.parquet")


# Read in the Parquet file created above. Parquet files are self-describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
parquetFile = sqlContext.read.parquet("people.parquet")


# Parquet files can also be registered as tables and then used in SQL statements.
parquetFile.registerTempTable("parquetFile");
teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenNames = teenagers.map(lambda p: "Name: " + p.name)
for teenName in teenNames.collect():
  print(teenName)



R:
--------------
# sqlContext from the previous example is used in this example.

schemaPeople # The DataFrame from the previous example.


# DataFrames can be saved as Parquet files, maintaining the schema information.
saveAsParquetFile(schemaPeople, "people.parquet")


# Read in the Parquet file created above. Parquet files are self-describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
parquetFile <- parquetFile(sqlContext, "people.parquet")


# Parquet files can also be registered as tables and then used in SQL statements.
registerTempTable(parquetFile, "parquetFile");
teenagers <- sql(sqlContext, "SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenNames <- map(teenagers, function(p) { paste("Name:", p$name)})
for (teenName in collect(teenNames)) {
  cat(teenName, "\n")
}

spark_1.x_sql_examples_1

----------------------------------------------------------------------------------
SPARK SQL INSTALLATION STEPS
----------------------------------------------------------------------------------

1. copy "kalyan_spark_jars" foder to "$SPARK_HOME" folder

2. copy "$SPARK_HOME/conf/spark-env.sh.template" file as "$SPARK_HOME/conf/spark-env.sh"

3. add the below line to "$SPARK_HOME/conf/spark-env.sh" file

export SPARK_CLASSPATH="$(echo $SPARK_HOME/kalyan_spark_jars/*.jar | tr ' ' ':')"

4. copy "$SPARK_HOME/conf/spark-defaults.conf.template" file as "$SPARK_HOME/conf/spark-defaults.conf"

5. add the below line to "$SPARK_HOME/conf/spark-defaults.conf" file

spark.driver.memory 5g

6. start the `spark-shell` with below command

$SPARK_HOME/bin/spark-shell

7. copy "input" foder to "/home/orienit/spark" folder


----------------------------------------------------------------------------------
SPARK SQL EXAMPLES with SCALA
----------------------------------------------------------------------------------
import org.apache.spark._
import org.apache.spark.sql._

val conf: SparkConf = new SparkConf().setAppName("Kalyan Sql Practice").setMaster("local[*]")
val sc: SparkContext = new SparkContext(conf)
val sqlContext: SQLContext = new org.apache.spark.sql.SQLContext(sc)


val df = sqlContext.read.json("file:///home/orienit/spark/input/student.json")
val df = sqlContext.read.parquet("file:///home/orienit/spark/input/student.parquet")

scala> df.show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
| spark|  1|  anil|2016|
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
| spark|  3|   raj|2016|
|hadoop|  4| sunil|2015|
| spark|  2|venkat|2016|
+------+---+------+----+


scala> df.select("name", "id").show
scala> df.select($"name", $"id").show
+------+---+
|  name| id|
+------+---+
|  anil|  1|
|anvith|  5|
|   dev|  6|
|   raj|  3|
| sunil|  4|
|venkat|  2|
+------+---+


scala> df.select(df("name"), df("id") + 1).show
scala> df.select($"name", $"id" + 1).show
+------+--------+
|  name|(id + 1)|
+------+--------+
|  anil|       2|
|anvith|       6|
|   dev|       7|
|   raj|       4|
| sunil|       5|
|venkat|       3|
+------+--------+

scala> df.filter(df("id") > 4).show
scala> df.filter($"id" > 4).show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
+------+---+------+----+


scala> df.filter(df("id") > 4 && df("id") < 6).show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
|hadoop|  5|anvith|2015|
+------+---+------+----+


scala> df.where(df("id") > 2).show
+------+---+------+----+
|  name| id|course|year|
+------+---+------+----+
|anvith|  5|hadoop|2015|
|   dev|  6|hadoop|2015|
|   raj|  3| spark|2016|
| sunil|  4|hadoop|2015|
+------+---+------+----+


scala> df.limit(4).show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
| spark|  1|  anil|2016|
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
| spark|  3|   raj|2016|
+------+---+------+----+


scala> df.toJSON.foreach(println)
{"name":"anil","id":1,"course":"spark","year":2016}
{"name":"anvith","id":5,"course":"hadoop","year":2015}
{"name":"dev","id":6,"course":"hadoop","year":2015}
{"name":"raj","id":3,"course":"spark","year":2016}
{"name":"sunil","id":4,"course":"hadoop","year":2015}
{"name":"venkat","id":2,"course":"spark","year":2016}


scala> df.groupBy("course","year").count.show
+------+----+-----+
|course|year|count|
+------+----+-----+
| spark|2016|    3|
|hadoop|2015|    3|
+------+----+-----+


scala> df.groupBy("id","course","year").count.show
+---+------+----+-----+
| id|course|year|count|
+---+------+----+-----+
|  6|hadoop|2015|    1|
|  5|hadoop|2015|    1|
|  3| spark|2016|    1|
|  4|hadoop|2015|    1|
|  2| spark|2016|    1|
|  1| spark|2016|    1|
+---+------+----+-----+


scala> df.orderBy($"course").show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
|hadoop|  4| sunil|2015|
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
| spark|  2|venkat|2016|
| spark|  3|   raj|2016|
| spark|  1|  anil|2016|
+------+---+------+----+


scala> df.orderBy($"course", $"name").show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
|hadoop|  4| sunil|2015|
| spark|  1|  anil|2016|
| spark|  3|   raj|2016|
| spark|  2|venkat|2016|
+------+---+------+----+


scala> df.sort("course", "id").show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
|hadoop|  4| sunil|2015|
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
| spark|  1|  anil|2016|
| spark|  2|venkat|2016|
| spark|  3|   raj|2016|
+------+---+------+----+




case class Contact(cid: Int, name: String, loc: String, pincode:Int)
case class Orders(oid: Int, cid: Int, status: String)

val contact = sc.textFile("file:///home/orienit/spark/input/contact.csv").map(_.split(","))
val cdf = contact.map(c => Contact(c(0).trim.toInt, c(1), c(2), c(3).trim.toInt)).toDF()

val orders = sc.textFile("file:///home/orienit/spark/input/orders.tsv").map(_.split("\t"))
val odf = orders.map(x => Orders(x(0).trim.toInt, x(1).trim.toInt, x(2))).toDF()

scala> cdf.show
+---+------+----+-------+
|cid|  name| loc|pincode|
+---+------+----+-------+
|  1|kalyan| hyd| 500072|
|  2|venkat| hyd| 500073|
|  3|prasad|bang| 600076|
|  4|anvith|bang| 600075|
+---+------+----+-------+

scala> odf.show
+---+---+-------+
|oid|cid| status|
+---+---+-------+
|111|  1|success|
|112|  1|failure|
|113|  2|success|
|114|  3|success|
|115|  2|failure|
+---+---+-------+

scala> cdf.join(odf).show
+---+------+----+-------+---+---+-------+
|cid|  name| loc|pincode|oid|cid| status|
+---+------+----+-------+---+---+-------+
|  1|kalyan| hyd| 500072|111|  1|success|
|  1|kalyan| hyd| 500072|112|  1|failure|
|  1|kalyan| hyd| 500072|113|  2|success|
|  2|venkat| hyd| 500073|111|  1|success|
|  2|venkat| hyd| 500073|112|  1|failure|
|  2|venkat| hyd| 500073|113|  2|success|
|  3|prasad|bang| 600076|111|  1|success|
|  3|prasad|bang| 600076|112|  1|failure|
|  3|prasad|bang| 600076|113|  2|success|
|  1|kalyan| hyd| 500072|114|  3|success|
|  1|kalyan| hyd| 500072|115|  2|failure|
|  2|venkat| hyd| 500073|114|  3|success|
|  2|venkat| hyd| 500073|115|  2|failure|
|  3|prasad|bang| 600076|114|  3|success|
|  3|prasad|bang| 600076|115|  2|failure|
|  4|anvith|bang| 600075|111|  1|success|
|  4|anvith|bang| 600075|112|  1|failure|
|  4|anvith|bang| 600075|113|  2|success|
|  4|anvith|bang| 600075|114|  3|success|
|  4|anvith|bang| 600075|115|  2|failure|
+---+------+----+-------+---+---+-------+


scala> cdf.join(odf, cdf("cid") === odf("cid")).show
+---+------+----+-------+---+---+-------+
|cid|  name| loc|pincode|oid|cid| status|
+---+------+----+-------+---+---+-------+
|  1|kalyan| hyd| 500072|111|  1|success|
|  1|kalyan| hyd| 500072|112|  1|failure|
|  2|venkat| hyd| 500073|113|  2|success|
|  2|venkat| hyd| 500073|115|  2|failure|
|  3|prasad|bang| 600076|114|  3|success|
+---+------+----+-------+---+---+-------+

scala> cdf.join(odf, cdf("cid") === odf("cid"), "left_outer").show
+---+------+----+-------+----+----+-------+
|cid|  name| loc|pincode| oid| cid| status|
+---+------+----+-------+----+----+-------+
|  1|kalyan| hyd| 500072| 111|   1|success|
|  1|kalyan| hyd| 500072| 112|   1|failure|
|  2|venkat| hyd| 500073| 113|   2|success|
|  2|venkat| hyd| 500073| 115|   2|failure|
|  3|prasad|bang| 600076| 114|   3|success|
|  4|anvith|bang| 600075|null|null|   null|
+---+------+----+-------+----+----+-------+

scala> cdf.join(odf, cdf("cid") === odf("cid"), "right_outer").show
+---+------+----+-------+---+---+-------+
|cid|  name| loc|pincode|oid|cid| status|
+---+------+----+-------+---+---+-------+
|  1|kalyan| hyd| 500072|111|  1|success|
|  1|kalyan| hyd| 500072|112|  1|failure|
|  2|venkat| hyd| 500073|113|  2|success|
|  2|venkat| hyd| 500073|115|  2|failure|
|  3|prasad|bang| 600076|114|  3|success|
+---+------+----+-------+---+---+-------+


scala> cdf.join(odf, cdf("cid") === odf("cid"), "full_outer").show
+---+------+----+-------+----+----+-------+
|cid|  name| loc|pincode| oid| cid| status|
+---+------+----+-------+----+----+-------+
|  1|kalyan| hyd| 500072| 111|   1|success|
|  1|kalyan| hyd| 500072| 112|   1|failure|
|  2|venkat| hyd| 500073| 113|   2|success|
|  2|venkat| hyd| 500073| 115|   2|failure|
|  3|prasad|bang| 600076| 114|   3|success|
|  4|anvith|bang| 600075|null|null|   null|
+---+------+----+-------+----+----+-------+


scala> cdf.join(odf, cdf("cid") === odf("cid"), "inner").show
+---+------+----+-------+---+---+-------+
|cid|  name| loc|pincode|oid|cid| status|
+---+------+----+-------+---+---+-------+
|  1|kalyan| hyd| 500072|111|  1|success|
|  1|kalyan| hyd| 500072|112|  1|failure|
|  2|venkat| hyd| 500073|113|  2|success|
|  2|venkat| hyd| 500073|115|  2|failure|
|  3|prasad|bang| 600076|114|  3|success|
+---+------+----+-------+---+---+-------+


scala> cdf.unionAll(cdf).show
+---+------+----+-------+
|cid|  name| loc|pincode|
+---+------+----+-------+
|  1|kalyan| hyd| 500072|
|  2|venkat| hyd| 500073|
|  3|prasad|bang| 600076|
|  4|anvith|bang| 600075|
|  1|kalyan| hyd| 500072|
|  2|venkat| hyd| 500073|
|  3|prasad|bang| 600076|
|  4|anvith|bang| 600075|
+---+------+----+-------+


scala> cdf.unionAll(df).show
+------+------+------+-------+
|   cid|  name|   loc|pincode|
+------+------+------+-------+
|     1|kalyan|   hyd| 500072|
|     2|venkat|   hyd| 500073|
|     3|prasad|  bang| 600076|
|     4|anvith|  bang| 600075|
| spark|     1|  anil|   2016|
|hadoop|     5|anvith|   2015|
|hadoop|     6|   dev|   2015|
| spark|     3|   raj|   2016|
|hadoop|     4| sunil|   2015|
| spark|     2|venkat|   2016|
+------+------+------+-------+


scala> cdf.intersect(cdf).show
+---+------+----+-------+
|cid|  name| loc|pincode|
+---+------+----+-------+
|  2|venkat| hyd| 500073|
|  4|anvith|bang| 600075|
|  1|kalyan| hyd| 500072|
|  3|prasad|bang| 600076|
+---+------+----+-------+


scala> cdf.intersect(df).show
+---+----+---+-------+
|cid|name|loc|pincode|
+---+----+---+-------+
+---+----+---+-------+


----------------------------------------------------------------------------------
RDBMS SPARK EXAMPLES
----------------------------------------------------------------------------------


Using JDBC To Connect Databases
----------------------------------------------------------------------------------
SPARK_CLASSPATH=<path-to-mysql-jar>/mysql-connector-java-5.1.34-bin.jar $SPARK_HOME/bin/spark-shell

$SPARK_HOME/bin/spark-shell --driver-class-path <path-to-mysql-jar>/mysql-connector-java-5.1.34-bin.jar

$SPARK_HOME/bin/spark-shell --jars <path-to-mysql-jar>/mysql-connector-java-5.1.34-bin.jar


Mysql Operations
----------------------------------------------------------------------------------
mysql -u root -p

CREATE DATABASE IF NOT EXISTS kalyan;

CREATE TABLE kalyan.student(name VARCHAR(50) PRIMARY KEY, id INT, course VARCHAR(50), year INT);

INSERT INTO kalyan.student(name, id, course, year) VALUES ('anil', 1, 'spark', 2016);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('venkat', 2, 'spark', 2016);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('raj', 3, 'spark', 2016);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('sunil', 4, 'hadoop', 2015);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('anvith', 5, 'hadoop', 2015);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('dev', 6, 'hadoop', 2015);

SELECT * FROM kalyan.student;

----------------------------------------------------------------------------------
To connect any RDBMS:
--------------------------
1. connection url
2. user name & password
3. driver class name
4. client jar
--------------------------

val jdbcDF = sqlContext.read.format("jdbc").options(
  Map("url" -> "jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop", "driver" -> "com.mysql.jdbc.Driver", "dbtable" -> "student")).load()

(or)

val jdbcDF = sqlContext.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop").option("driver", "com.mysql.jdbc.Driver").option("dbtable", "student").load()

(or)

val jdbcDF = sqlContext.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/kalyan").option("driver", "com.mysql.jdbc.Driver").option("dbtable", "student").option("user", "root").option("password", "hadoop").load()


(or)

val prop = new java.util.Properties

val jdbcDF = sqlContext.read.jdbc("jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop", "student", prop)


(or)

val prop = new java.util.Properties
prop.setProperty("driver","com.mysql.jdbc.Driver")
prop.setProperty("user","root")
prop.setProperty("password","hadoop")

val jdbcDF = sqlContext.read.jdbc("jdbc:mysql://localhost:3306/kalyan", "student", prop)


(or)

val jdbcDF = sqlContext.read.jdbc("jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop", "student", Array("course='spark'"), prop)

jdbcDF.show()

val jdbcDF = sqlContext.read.jdbc("jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop", "student", Array("course='spark'", "year=2015"), prop)

jdbcDF.show()

val userdata = jdbcDF.select("name", "id")
userdata.show()


----------------------------------------------------------------------------------
Saving the data from `dataframe` to `other systems`
----------------------------------------------------------------------------------

val prop = new java.util.Properties
prop.setProperty("driver","com.mysql.jdbc.Driver")
prop.setProperty("user","root")
prop.setProperty("password","hadoop")

val jdbcDF = sqlContext.read.jdbc("jdbc:mysql://localhost:3306/kalyan", "student", prop)

jdbcDF.show()

----------------------------------------------------------------------------------

jdbcDF.save("file:///home/orienit/spark/output/student_json", "json")

jdbcDF.save("file:///home/orienit/spark/output/student_json", "json", SaveMode.Overwrite)

jdbcDF.save("file:///home/orienit/spark/output/student_json", "json", SaveMode.Append)

----------------------------------------------------------------------------------

jdbcDF.save("file:///home/orienit/spark/output/student_orc", "orc")

jdbcDF.save("file:///home/orienit/spark/output/student_orc", "orc", SaveMode.Overwrite)

jdbcDF.save("file:///home/orienit/spark/output/student_orc", "orc", SaveMode.Append)

----------------------------------------------------------------------------------

jdbcDF.save("file:///home/orienit/spark/output/student_parquet")

jdbcDF.save("file:///home/orienit/spark/output/student_parquet", SaveMode.Overwrite)

jdbcDF.save("file:///home/orienit/spark/output/student_parquet", SaveMode.Append)

----------------------------------------------------------------------------------

jdbcDF.save("file:///home/orienit/spark/output/student_parquet", "parquet")

jdbcDF.save("file:///home/orienit/spark/output/student_parquet", "parquet", SaveMode.Overwrite)

jdbcDF.save("file:///home/orienit/spark/output/student_parquet", "parquet", SaveMode.Append)

jdbcDF.saveAsParquetFile("file:///home/orienit/spark/output/student_parquet_1")

----------------------------------------------------------------------------------

val prop = new java.util.Properties
prop.setProperty("driver","com.mysql.jdbc.Driver")
prop.setProperty("user","root")
prop.setProperty("password","hadoop")

jdbcDF.write.jdbc("jdbc:mysql://localhost:3306/kalyan", "student1", prop)

jdbcDF.write.mode("overwrite").jdbc("jdbc:mysql://localhost:3306/kalyan", "student1", prop)

jdbcDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/kalyan", "student1", prop)

jdbcDF.insertIntoJDBC("jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop", "student1", false)

jdbcDF.insertIntoJDBC("jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop", "student1", true)

jdbcDF.createJDBCTable("jdbc:mysql://localhost:3306/kalyan?user=root&password=hadoop", "student2", true)

----------------------------------------------------------------------------------

import org.apache.spark.sql._

jdbcDF.saveAsTable("student")

jdbcDF.saveAsTable("student", SaveMode.Overwrite)

jdbcDF.saveAsTable("student", SaveMode.Append)

jdbcDF.saveAsTable("kalyan.student")

jdbcDF.insertInto("kalyan.student")

----------------------------------------------------------------------------------
Creating udfs in spark
----------------------------------------------------------------------------------

val students1 = List(("rajesh", 11, "spark" , 2016), ("nagesh", 12, "spark" , 2016), ("ganesh", 13, "spark" , 2016))

val studentsRdd1 = sc.parallelize(students1)


case class Student(name: String, id: Int, course: String, year:Int)

val students2 = students1.map( t => Student(t._1, t._2, t._3, t._4))

val studentsRdd2 = sc.parallelize(students2)


val df1 = sqlContext.createDataFrame(studentsRdd1)
df1.registerTempTable("student1")

val df2 = sqlContext.createDataFrame(studentsRdd2)
df2.registerTempTable("student2")


scala> sqlContext.sql("select * from student1").show
+------+---+-----+----+
|    _1| _2|   _3|  _4|
+------+---+-----+----+
|rajesh| 11|spark|2016|
|nagesh| 12|spark|2016|
|ganesh| 13|spark|2016|
+------+---+-----+----+


scala> sqlContext.sql("select * from student2").show
+------+---+------+----+
|  name| id|course|year|
+------+---+------+----+
|rajesh| 11| spark|2016|
|nagesh| 12| spark|2016|
|ganesh| 13| spark|2016|
+------+---+------+----+

scala> df1.select("_1", "_2").show
+------+---+
|    _1| _2|
+------+---+
|rajesh| 11|
|nagesh| 12|
|ganesh| 13|
+------+---+


scala> df2.select("name", "id").show
+------+---+
|  name| id|
+------+---+
|rajesh| 11|
|nagesh| 12|
|ganesh| 13|
+------+---+

----------------------------------------------------------------------------------

def lengthFunc(name: String) = { name.length }
sqlContext.udf.register("mylength", lengthFunc(_:String))

// Now we can use our function directly in SparkSQL.
sqlContext.sql("SELECT name, mylength(name) from student2").show
sqlContext.sql("SELECT name, mylength(name) as len from student2").show

// but not outside
df2.select($"name", mylength($"name"), $"id").show // fails


import org.apache.spark.sql.functions.udf
val lengthUdf = udf(lengthFunc(_:String))

// now this works
df2.select($"name", lengthUdf($"name"), $"id").show


----------------------------------------------------------------------------------
SPARK SQL with `DATA SETS`
----------------------------------------------------------------------------------

scala> val df = sc.makeRDD(1 to 10).toDF()
df: org.apache.spark.sql.DataFrame = [_1: int]

scala> df.map(_ + 1).collect()
<console>:30: error: type mismatch;
 found   : Int(1)
 required: String
              df.map(_ + 1).collect()

scala> df.map(row => row(0).toString.toInt + 1).collect
res83: Array[Int] = Array(2, 3, 4, 5, 6, 7, 8, 9, 10, 11)

scala> df.map(row => row.getAs[Int]("_1") + 1).collect
res97: Array[Int] = Array(2, 3, 4, 5, 6, 7, 8, 9, 10, 11)



scala> val ds = (1 to 10).toDS()
ds: org.apache.spark.sql.Dataset[Int] = [value: int]

scala> ds.map(_ + 1).collect()
res71: Array[Int] = Array(2, 3, 4, 5, 6, 7, 8, 9, 10, 11)

----------------------------------------------------------------------------------

case class Student(name: String, id: Long, course: String, year: Long)

val path = "file:///home/orienit/spark/input/student.json"

scala> val df = sqlContext.read.json(path)
df: org.apache.spark.sql.DataFrame = [course: string, id: bigint, name: string, year: bigint]


scala> val ds = sqlContext.read.json(path).as[Student]
ds: org.apache.spark.sql.Dataset[Student] = [name: string, id: bigint, course: string, year: bigint]


Note:-
1. DataFrames can be converted to a Dataset by providing a class.
2. Datasets can be converted to a DataFrame directly

scala> val ds = df.as[Student]
ds: org.apache.spark.sql.Dataset[Student] = [name: string, id: bigint, course: string, year: bigint]


scala> df.show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
| spark|  1|  anil|2016|
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
| spark|  3|   raj|2016|
|hadoop|  4| sunil|2015|
| spark|  2|venkat|2016|
+------+---+------+----+


scala> ds.show
+------+---+------+----+
|course| id|  name|year|
+------+---+------+----+
| spark|  1|  anil|2016|
|hadoop|  5|anvith|2015|
|hadoop|  6|   dev|2015|
| spark|  3|   raj|2016|
|hadoop|  4| sunil|2015|
| spark|  2|venkat|2016|
+------+---+------+----+


scala> ds.collect.foreach(println)
Student(anil,1,spark,2016)
Student(anvith,5,hadoop,2015)
Student(dev,6,hadoop,2015)
Student(raj,3,spark,2016)
Student(sunil,4,hadoop,2015)
Student(venkat,2,spark,2016)


scala> df.collect.foreach(println)
[spark,1,anil,2016]
[hadoop,5,anvith,2015]
[hadoop,6,dev,2015]
[spark,3,raj,2016]
[hadoop,4,sunil,2015]
[spark,2,venkat,2016]



----------------------------------------------------------------------------------
HIVE SPARK EXAMPLES
----------------------------------------------------------------------------------

import org.apache.spark.sql.hive.HiveContext

val hiveContext = new HiveContext(sc)

hiveContext.sql("CREATE DATABASE IF NOT EXISTS kalyan")
hiveContext.sql("CREATE TABLE IF NOT EXISTS kalyan.src (key INT, value STRING)")
hiveContext.sql("LOAD DATA LOCAL INPATH '${env:SPARK_HOME}/examples/src/main/resources/kv1.txt' INTO TABLE kalyan.src")

val countdata = hiveContext.sql("SELECT count(*) from kalyan.src")
countdata.collect()

val input = hiveContext.sql("FROM kalyan.src SELECT key, value")
val data = input.map(_.getInt(0))

println(input.collect().toList)
println(data.collect().toList)

input.collect().foreach(println)




----------------------------------------------------------------------------------
CASSANDRA SPARK EXAMPLES
----------------------------------------------------------------------------------

DROP KEYSPACE kalyan;

CREATE KEYSPACE kalyan WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1 };

CREATE TABLE kalyan.student(name text PRIMARY KEY, id int, course text, year int);

INSERT INTO kalyan.student(name, id, course, year) VALUES ('anil', 1, 'spark', 2016);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('venkat', 2, 'spark', 2016);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('raj', 3, 'spark', 2016);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('sunil', 4, 'hadoop', 2015);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('anvith', 5, 'hadoop', 2015);
INSERT INTO kalyan.student(name, id, course, year) VALUES ('dev', 6, 'hadoop', 2015);

SELECT * FROM kalyan.student;

----------------------------------------------------------------------------------

sc.stop()

import org.apache.spark._
import com.datastax.spark.connector._

val conf = new SparkConf(true).set("spark.cassandra.connection.host", "127.0.0.1").set("spark.cassandra.input.split.size_in_mb", "268435456")

val sc = new SparkContext("local[*]", "kalyan", conf)

val rdd = sc.cassandraTable("kalyan", "student")

println(rdd.count)

println(rdd.first)

rdd.groupBy(x => x.getInt("year")).foreach(println)

----------------------------------------------------------------------------------

val students = List(("rajesh", 11, "spark" , 2016), ("nagesh", 12, "spark" , 2016), ("ganesh", 13, "spark" , 2016))

val studentsRdd = sc.parallelize(students)

studentsRdd.saveToCassandra("kalyan", "student", SomeColumns("name", "id", "course", "year"))     

----------------------------------------------------------------------------------

import org.apache.spark.sql._
import org.apache.spark.sql.cassandra._

val csc = new CassandraSQLContext(sc)

val df = csc.sql("SELECT * from kalyan.student")
df.foreach(println)






----------------------------------------------------------------------------------
PHOENIX SPARK EXAMPLES
----------------------------------------------------------------------------------

CREATE TABLE STUDENT(name VARCHAR PRIMARY KEY, id INTEGER, course VARCHAR, year INTEGER);

UPSERT INTO STUDENT(name, id, course, year) VALUES ('anil', 1, 'spark', 2016);
UPSERT INTO STUDENT(name, id, course, year) VALUES ('venkat', 2, 'spark', 2016);
UPSERT INTO STUDENT(name, id, course, year) VALUES ('raj', 3, 'spark', 2016);
UPSERT INTO STUDENT(name, id, course, year) VALUES ('sunil', 4, 'hadoop', 2015);
UPSERT INTO STUDENT(name, id, course, year) VALUES ('anvith', 5, 'hadoop', 2015);
UPSERT INTO STUDENT(name, id, course, year) VALUES ('dev', 6, 'hadoop', 2015);

SELECT * FROM STUDENT;

----------------------------------------------------------------------------------

sc.stop()

import org.apache.spark._
import org.apache.spark.sql._
import org.apache.spark.sql.types._
import org.apache.spark.rdd.RDD
import org.apache.phoenix.spark._

val sc = new SparkContext("local", "kalyan")
val sqlContext = new SQLContext(sc)

val df = sqlContext.load(
  "org.apache.phoenix.spark",
  Map("table" -> "STUDENT", "zkUrl" -> "localhost:2181")
)

----------------------------------------------------------------------------------

df.show

df.select("NAME").show()

df.select(df("NAME")).show()

df.select("NAME", "ID").show()

df.select(df("NAME"), df("ID")).show()

df.filter(df("ID") === 1L).show()

df.filter(df("ID") >= 2L).show()

df.filter(df("COURSE") === "spark" && df("ID") >= 1L).show

df.filter(df("COURSE") === "spark" && df("ID") === 1L).select(df("ID")).show

--------------------------------------------------------------------------

import org.apache.hadoop.conf.Configuration

val configuration = new Configuration()

val df = sqlContext.phoenixTableAsDataFrame(
 "STUDENT", Seq[String](), zkUrl = Some("localhost:2181"), conf = configuration
)

df.show


val df = sqlContext.phoenixTableAsDataFrame(
 "STUDENT", Seq[String]("ID", "NAME"), zkUrl = Some("localhost:2181"), conf = configuration
)

df.show

--------------------------------------------------------------------------

val rdd = sc.phoenixTableAsRDD(
 "STUDENT", Seq[String](), zkUrl = Some("localhost:2181"), conf = configuration
)

rdd.foreach(println)

val rdd = sc.phoenixTableAsRDD(
 "STUDENT", Seq[String]("ID", "NAME"), zkUrl = Some("localhost:2181"), conf = configuration
)

rdd.foreach(println)

val firstId = rdd.first()("ID").asInstanceOf[Int]
val firstNAME = rdd.first()("NAME").asInstanceOf[String]

--------------------------------------------------------------------------

val students = List(("rajesh", 11, "spark" , 2016), ("nagesh", 12, "spark" , 2016), ("ganesh", 13, "spark" , 2016))

val studentsRdd = sc.parallelize(students)

studentsRdd.saveToPhoenix("STUDENT", Seq("NAME", "ID", "COURSE", "YEAR"), zkUrl = Some("localhost:2181"))

--------------------------------------------------------------------------

import org.apache.spark.sql._
import org.apache.spark.sql.types._

// Load from `STUDENT`
val df = sqlContext.load("org.apache.phoenix.spark", Map("table" -> "STUDENT", "zkUrl" -> "localhost:2181"))


// Save to `STUDENT1`
CREATE TABLE STUDENT1(name VARCHAR PRIMARY KEY, id INTEGER, course VARCHAR, year INTEGER);

df.save("org.apache.phoenix.spark", SaveMode.Overwrite, Map("table" -> "STUDENT1", "zkUrl" -> "localhost:2181"))

----------------------------------------------------------------------------------

SCHEMA_GENERATION_WRAPPER_SRIPT

#Source DBSRVR config file
#Taking user input and passing to DBSRVR file

echo $0
BIN_DIR=`dirname $0`
echo $BIN_DIR
source $BIN_DIR/cdl-bl-db-env.conf $2
#Source Hadoop Conf file
source $BIN_DIR/cdl-bl-env.conf

#Source the parameter file
echo $1
source $1

#Job Unix Log File Location
JobRunDate=`date +%Y:%m:%d:%H:%M:%S`
SummaryLogFileNm=SCHEMA_GENERATION_Wrapper_`date +%Y%m%d%H%M%S`_${TBL_NM}_$USER.log
SLogFullFileName=$CDL_LOGS/${SummaryLogFileNm}
exec 1> $SLogFullFileName 2>&1


echo  $CDL_CONF
echo  $CDL_PARAM
echo  $CDL_LOGS


#Source bash_profile to get credentials for Teradata BTEQ Script
source $HOME/.bash_profile

echo extracting columns started
sh $CDL_BIN/SCHEMA_DIR/EXTRACT_COLUMNS.sh $1 $2
if [ $? -eq 0 ]; then
echo extracting columns completed
else
echo extracting columns failed
exit 1
fi


echo avro schema genertion started
sh $CDL_BIN/SCHEMA_DIR/AVRO_SCHEMA_GENERATION.sh $1 $2
if [ $? -eq 0 ]; then
echo avro schema genertion completed
else
echo avro schema genertion failed
exit 1
fi


echo Parquet schema genertion started
sh $CDL_BIN/SCHEMA_DIR/PARQUET_SCHEMA_GENERATION.sh $1 $2
if [ $? -eq 0 ]; then
echo Parquet schema genertion completed
else
echo Parquet schema genertion failed
exit 1
fi

if [ "${ENVIRONMENT}" == "pr" ]
then
allphi_schema=${ENVIRONMENT}_${PROGRAM}${PROJECT_ABBREVATION}${PHI_NOPHI_DB}_r000_wh_allphi
else
allphi_schema=${ENVIRONMENT}_${PROGRAM}${PROJECT_ABBREVATION}${PHI_NOPHI_DB}_${HIVE_GBD_NOGBD}_r000_wh_allphi
fi

if [ "${PROJECT_ABBREVATION_AR}" == "pdlr_ar" ]
then
cdl_schema_name=${ENVIRONMENT}_${PROGRAM}${PROJECT_ABBREVATION}${PHI_NOPHI_DB}_${HIVE_GBD_NOGBD}_r000_ar
else
cdl_schema_name=${cdl_schema_name_wh}
fi
echo "allphi_schema: ${allphi_schema}"
echo "cdl_schema_name: ${cdl_schema_name}"

#Create view
beeline -u $BEELINE_CMD  --outputformat=tsv2 --showHeader=false hive -e "CREATE OR REPLACE VIEW ${allphi_schema}.${cdl_table_name} AS select * from ${cdl_schema_name}.${cdl_table_name};"
#Check Script successfully executed or not
if [ $? -eq 0 ]; then
    echo "Succefully create the view ${allphi_schema} "
else

echo "Beeline command  failed create the existing view  ${allphi_schema}"
#exit 1
fi

DL_PARQUET_WRAPPER_SRIPT

#====================================================================================================================================================
# Title            : CDL_PARQUET_Wrapper
# ProjectName      : CDL BaseLine Loads
# Filename         : CDL_PARQUET_Wrapper.sh
# Description      : Script moves data from Avro Hive external table into Parquet Hive external table
# Developer        :xxxxxx
# Created on       : AUG 2017
# Location         : xxxxxx
# Logic            :
# Parameters       : Parameter file name
# Return codes     :
# Date                         Ver#     Modified By(Name)                 Change and Reason for Change
# ----------    -----        -----------------------------               --------------------------------------
# 2017/07/13                      1     Initial Version

#  ***************************************************************************************************************************
######### Below scrtip performs the following
######### 1. Execute the Reconciliation -1 : Compare source record counts to the avro table count. If the count matches , move the data to parquet table.
#########     if the counts do not match, abort the job .
######### 2. Insert data into Parquet table
######### 3. Write LoadstartDate And LoadEndDate in param file

echo "Execution of  parquet  Schema  Creation"
#Source DBSRVR config file
#Taking user input and passing to DBSRVR file
db_conf=$2
echo "DB Conf value"
echo $db_conf
echo $0
BIN_DIR=`dirname $0`
echo $BIN_DIR
echo "Sourcing .env file"
source $BIN_DIR/cdl-bl-db-env.conf $db_conf
#Source Hadoop config file
source $BIN_DIR/cdl-bl-env.conf

#Source the parameter file
echo $1
source $1

echo  $CDL_CONF
echo  $CDL_PARAM
echo  $CDL_LOGS


#Source bash_profile to get credentials for Teradata BTEQ Script
#source ~/.bash_profile

SummaryLogFileNm=CDL_AVRO_PARQUET_`date +%Y%m%d%H%M%S`_${TBL_NM}_$USER.log
SLogFullFileName=$CDL_LOGS/$SummaryLogFileNm
exec 1> $SLogFullFileName 2>&1


#Creating Parquet Table before inserting data.

 #sh CDL_PARQUET_TABLE_CREATION_WRAPPER.sh

### Variable Declarations
error_count=0
YYYY=`date +%Y`
MM=`date +%m`
RunDate=`date +%Y:%m:%d:%H:%M:%S`


fnLogMsg()
{
     LogTime=`date +%Y%m%d%H%M%S`
     LogMsg="$LogTime:$1:$2"
     echo $LogMsg >> $SLogFullFileName
     echo $LogMsg
}

fnStatusCheck()
{
     CommandRC="$1"
     SuccessMSG="$2"
     FailureMSG="$3"
             
    if [ $CommandRC -eq 0 ]
    then
        fnLogMsg INFO "${SuccessMSG}"
        else
            fnLogMsg INFO "${FailureMSG}"
        exit 1
        error_count =`expr $error_count + 1`
    fi
}



#Source bash_profile to get credentials for Teradata BTEQ Script

#source ./.bash_profile

#echo "Logon server name"

#echo $LOGON

#Check  if file is load_log file.If yes skip the recon process



#########Reconciliation -1 : Compare source record counts to the avro table count. If the count matches , move the data to parquet table.
######### if the counts do not match, abort the job .

#Create directory if not exist
dir=$CDL_LOGS/Reconciliation
if [[ ! -e $dir ]]; then
    mkdir $dir
elif [[  -d $dir ]]; then
    echo "$dir already exist"
fi

#Remove Temp Files if exist
AvroReconFile=$CDL_LOGS/Reconciliation/avro_count_recon_${TBL_NM}_$RUN_NB.txt
if [ -f "$AvroReconFile" ]
then
    rm $AvroReconFile

   echo "AvroReconFile Temp File deleted"
fi

#Remove Temp Files if exist
TeradataReconFile=$CDL_LOGS/Reconciliation/Teradata_count_recon_${TBL_NM}_$RUN_NB.txt
if [ -f "$TeradataReconFile" ]
then
    rm $TeradataReconFile

   echo "TeradataReconFile Temp File deleted"
fi


#GET Avro Table Count

#impala-shell -k --ssl -i $IMPALA_CMD -B  -q  "REFRESH ${cdl_schema_name_sg}.${cdl_incoming_table_name};SELECT count(*) from  ${cdl_schema_name_sg}.${cdl_incoming_table_name};" --output_file=$CDL_LOGS/Reconciliation/avro_count_recon_${TBL_NM}_$RUN_NB.txt

beeline -u $BEELINE_CMD  --outputformat=tsv2 --showHeader=false -e "SELECT count(*) from  ${cdl_schema_name_sg}.${cdl_incoming_table_name};">>$CDL_LOGS/Reconciliation/avro_count_recon_${TBL_NM}_$RUN_NB.txt

if [ $? -gt 0 ]; then
echo "Error in getting count from Avro external table"
exit 1
fi


cat $CDL_LOGS/CDL_BTEQ_RECON_${TBL_NM}_$USER.log|grep 'Count' -A 2|tail -1 >>$CDL_LOGS/Reconciliation/Teradata_count_recon_${TBL_NM}_$RUN_NB.txt



#Getting the counts for avro and teradata
avroRecon_count=`cat $CDL_LOGS/Reconciliation/avro_count_recon_${TBL_NM}_$RUN_NB.txt`
TeradataRecon_count=`cat $CDL_LOGS/Reconciliation/Teradata_count_recon_${TBL_NM}_$RUN_NB.txt`


echo "avro file count =" $avroRecon_count
echo "Teradata source count =" $TeradataRecon_count
######Compare Teradata  count with Avro count. If the counts match , then proceed with next  steps. Else exit the code.

check_load_log_file=` echo $1 | grep 'load_log'`
if [  -z "$check_load_log_file" ]; then
echo "File is not load_log file.Teradata Recon required"

if [ $avroRecon_count -eq $TeradataRecon_count ]
then

echo "Teradata Avro Counts are matching"

else

if [ "$STAGE_TABLE_NEEDED" == "YES" ]
then
echo "Teradata Recon count and avro count are not matched"
exit 1
else
echo "Teradata Recon count and avro count are not matched"
echo "Teradata Recon count and avro count are not matched. But, Process continues to load the Data in to Hive" |mail -s "Teradata Recon count and avro count are not matched for ${TBL_NM}" dl-cdl_fndtn_lyr@anthem.com
#exit 1
fi
fi
else
echo "load_log file.Skipping Recon process"
echo $check_load_log_file

fi
#Export avro table count for audit wrapper
avroRecon_cnt=`echo ${avroRecon_count//[[:blank:]]/}`
echo export avroRecon_count=$avroRecon_cnt >>$1



################################### STEP -2 : Move the data from Avro table into Parquet table
echo "STEP -1 : Move the data from Avro table into Parquet table"

### Execution of script starts here
echo "Values required for Parquet insert"
echo $cdl_incoming_table_name
echo $cdl_table_name
echo $cdl_schema_name_wh
echo $is_partition


#Table name for hive query
        tbl_nm_hive="'$cdl_table_name'"
        echo "hive query table name"
        echo $tbl_nm_hive

############ Get the updated Load Log Key-CDH_LOAD_LOG



#Deleting Temp File

CDLFile=$CDL_LOGS/cdl_count_$TBL_NM.txt
if [ -f "$CDLFile" ]
then
     rm $CDLFile

   echo "CDLFile Temp File deleted"
fi


# Table name in CDH_LOD_LOG query to get max(CDH_LOAD_LOG) value
teradata_table_name=$TBL_NM
teradata_table_name_for_hive="'$teradata_table_name'"
echo "Hive query table name"
echo $teradata_table_name_for_hive
process_nm_param=$PROCESS_NM
process_nm_hive="'$process_nm_param'"
echo "Process Name"
echo $process_nm_hive
subj_area_param=$SUBJ_AREA_NM
subj_area_hive="'$subj_area_param'"
echo "subj_area_hive Name"
echo $subj_area_hive
#connectimg beeline to get the max $CDH_LOAD_LOG_KEY
echo "Schema name"
echo $cdl_schema_name_sg
echo  beeline -u $BEELINE_CMD --outputformat=tsv2 --showHeader=false -e "set hive.exec.max.dynamic.partitions.pernode=500;set hive.auto.convert.join=false;set hive.exec.dynamic.partition=true; set hive.exec.dynamic.partition.mode=nonstrict; select max(CDH_LOAD_LOG_KEY) FROM $cdl_schema_name_sg.CDH_LOAD_LOG where SUB_PROCESS_NM=$teradata_table_name_for_hive and SUBJ_AREA_NM=$subj_area_hive and PROCESS_NM=$process_nm_hive limit 1;"



#beeline -u $BEELINE_CMD --outputformat=tsv2 --showHeader=false -e "set hive.exec.max.dynamic.partitions.pernode=500;set hive.auto.convert.join=false;set hive.exec.dynamic.partition=true; set hive.exec.dynamic.partition.mode=nonstrict; select max(CDH_LOAD_LOG_KEY) FROM $cdl_schema_name_sg.CDH_LOAD_LOG where SUB_PROCESS_NM=$teradata_table_name_for_hive and SUBJ_AREA_NM=$subj_area_hive and PROCESS_NM=$process_nm_hive limit 1;" > $CDL_LOGS/cdl_count_$TBL_NM.txt


#Check load_log_key-If NULL then set to 1 otherwise load_log_key+1
#load_log_key=`cat $CDL_LOGS/cdl_count_$TBL_NM.txt`
#echo "beeline result"
#echo $load_log_key
#First entry of table in Audit table
#if [ "$load_log_key" == "NULL" ]; then
#CDH_LOAD_LOG_KEY=1
#echo $CDH_LOAD_LOG_KEY
#  echo "load_log_key is null -initializing to 1"
#else
#CDH_LOAD_LOG_KEY=`expr $load_log_key + 1`
CDH_LOAD_LOG_KEY=`java -jar $CDL_BIN/Unique_number/UniqueNumber.jar`
#fi

#Checking Table Partition
#LoadStartDate=`date +%Y-%m-%d %H:%M:%S`
#Exporting For Audit table
LoadStartDate=`date +%Y:%m:%d:%H:%M:%S`
echo "LoadStartDate******************"
echo $LoadStartDate

echo "Writing variables in file"
echo $1
#echo export CDH_LOAD_LOG_KEY=$CDH_LOAD_LOG_KEY >>$1
echo export CDH_LOAD_LOG_KEY=$CDH_LOAD_LOG_KEY >>$1
echo export LoadStartDate=$LoadStartDate >>$1




set hive.execution.engine=mapreduce

echo CDH_LOAD_LOG_KEY : $CDH_LOAD_LOG_KEY
echo RUN_NB: $RUN_NB

#Check the RUN_NB is param file and if RUN_NB=1 then delete the old data parquet table data
if [ $RUN_NB -eq 1 ]
then
    # hadoop fs -rmr -skipTrash $PARQUET_LOCATION
hadoop fs -rmr -skipTrash $PARQUET_LOCATION/*
   echo "Parquet old files deleted"
   #Initialize Table Count To 0
   PQT_ROWS_LOADED_NB=0
   echo "Count of Parquet Table "
   echo $PQT_ROWS_LOADED_NB
fi

#Create directory if not exist
audit_dir=$CDL_TEMP_AUDIT/Audit
if [[ ! -e $audit_dir ]]; then
    mkdir $audit_dir
elif [[  -d $audit_dir ]]; then
    echo "$audit_dir already exist"
fi

ParquetFile=$audit_dir/parquet_count_${TBL_NM}_$USER.txt

if [ -f "$ParquetFile" ]
then
    rm $ParquetFile

   echo "Parquet Temp File deleted"
fi

#Get the parquet table name
parquet_table_name=$cdl_table_name

#Check parquet count before inserting data to avoid duplication
if [ $RUN_NB != 1 ]

then

echo "RUN_NB is not 1.Check Parquet Table Count"
#Get the parquet table count
#echo beeline -u $BEELINE_CMD  --outputformat=tsv2 --showHeader=false -e "SELECT count(*) from  ${cdl_schema_name_wh}.${parquet_table_name} where ${PARQUET_RECON_CON};" >> $audit_dir/parquet_count_$TBL_NM_$USER.txt

beeline -u $BEELINE_CMD  --outputformat=tsv2 --showHeader=false -e "SELECT count(*) from  ${cdl_schema_name_wh}.${parquet_table_name} where ${PARQUET_RECON_CON};" >> $audit_dir/parquet_count_${TBL_NM}_$USER.txt

#impala-shell -k --ssl -i $IMPALA_CMD -B  -q  "invalidate metadata ${cdl_schema_name_wh}.${parquet_table_name} ;SELECT count(*) from  ${cdl_schema_name_wh}.${parquet_table_name} where ${PARQUET_RECON_CON};" --output_file=$audit_dir/parquet_count_$TBL_NM_$USER.txt

#Check Script successfully executed or not
if [ $? -eq 0 ]; then
    echo "Succefully get count from  Parquet external table "
else

echo "Beeline command to get count from  Parquet external table failed"
exit 1
fi



#Getting the counts for avro and parquet
echo "Paquet Count from Param file"
parquet_count=`cat $audit_dir/parquet_count_${TBL_NM}_$USER.txt`

echo $parquet_count

#Parquet table count
PQT_ROWS_LOADED_NB=`echo ${parquet_count//[[:blank:]]/}`

echo $PQT_ROWS_LOADED_NB

#End of RUN_NB if condition
fi


#Compare the counts before deleting the temp files and tables
if [ "$PQT_ROWS_LOADED_NB" == "0" ]
then


#SPARK2.2

echo spark2-submit --verbose --class com.anthem.cdl.preprocess.CDL_DDL_Insertion --master yarn --deploy-mode cluster --queue ${QUEUE_NAME} --driver-memory 8G --executor-memory 30G --executor-cores 4 --num-executors 40 --conf spark.driver.maxResultSize=3g --conf spark.yarn.executor.memoryOverhead=8192 --conf spark.yarn.driver.memoryOverhead=8192 --conf spark.network.timeout=600 --conf hive.execution.engine=spark --conf spark.port.maxRetries=20 --conf spark.sql.broadcastTimeout=4800 --conf spark.executor.heartbeatInterval=30s --conf spark.dynamicAllocation.initialExecutors=10 --conf spark.dynamicAllocation.minExecutors=10 --conf spark.dynamicAllocation.maxExecutors=40 --files /etc/spark2/conf.cloudera.spark2_on_yarn/yarn-conf/hive-site.xml --jars /opt/cloudera/parcels/CDH-5.12.2-1.cdh5.12.2.p0.4/jars/hive-contrib-1.1.0-cdh5.12.2.jar,/usr/lib/tdch/1.5/lib/terajdbc4.jar,/usr/lib/tdch/1.5/lib/tdgssconfig.jar --name $TBL_NM $BIN_DIR/$SPARK_JAR $PARQUET_TEMP_LOCATION/final_parquet_data_$cdl_table_name.txt $cdl_schema_name_wh $cdl_schema_name_sg $cdl_table_name $cdl_incoming_table_name $PARQUET_LOCATION $CDH_LOAD_LOG_KEY $PARTITION_COLUMN $PARTITION_COLUMN_1

echo insert to parquet from avro started

export JAVA_HOME=/usr/java/latest

spark2-submit --verbose --class com.anthem.cdl.preprocess.CDL_DDL_Insertion --master yarn --deploy-mode cluster --queue ${QUEUE_NAME} --driver-memory 8G --executor-memory 30G --executor-cores 4 --num-executors 40 --conf spark.driver.maxResultSize=3g --conf spark.yarn.executor.memoryOverhead=8192 --conf spark.yarn.driver.memoryOverhead=8192 --conf spark.network.timeout=600 --conf hive.execution.engine=spark --conf spark.port.maxRetries=20 --conf spark.sql.broadcastTimeout=4800 --conf spark.executor.heartbeatInterval=30s --conf spark.dynamicAllocation.initialExecutors=10 --conf spark.dynamicAllocation.minExecutors=10 --conf spark.dynamicAllocation.maxExecutors=40 --files /etc/spark2/conf.cloudera.spark2_on_yarn/yarn-conf/hive-site.xml --jars /opt/cloudera/parcels/CDH-5.12.2-1.cdh5.12.2.p0.4/jars/hive-contrib-1.1.0-cdh5.12.2.jar,/usr/lib/tdch/1.5/lib/terajdbc4.jar,/usr/lib/tdch/1.5/lib/tdgssconfig.jar --name $TBL_NM $BIN_DIR/$SPARK_JAR $PARQUET_TEMP_LOCATION/final_parquet_data_$cdl_table_name.txt $cdl_schema_name_wh $cdl_schema_name_sg $cdl_table_name $cdl_incoming_table_name $PARQUET_LOCATION $CDH_LOAD_LOG_KEY $PARTITION_COLUMN $PARTITION_COLUMN_1


#Get spark application URL
 application_url=`grep tracking  $CDL_LOGS/$SummaryLogFileNm|head -1`

 #Extract application_id from URL
application_id=$(echo $application_url | sed 's:/*$::')

#Get application name
application_name=`echo $application_id|rev | cut -d'/' -f 1 | rev`

#Remove temp files
rm $CDL_LOGS/temp_app_details_${TBL_NM}_$USER.txt

#Get application status details and save in temp file
yarn application --status $application_name >$CDL_LOGS/temp_app_details_${TBL_NM}_$USER.txt

#Get the application final status
app_status=`grep Final-State $CDL_LOGS/temp_app_details_${TBL_NM}_$USER.txt`
final_app_status=`echo $app_status|rev | cut -d':' -f 1 | rev|tail -1`

#echo $final_app_status
status="SUCCEEDED"

echo $final_app_status
echo $status
#Compare application status
if [ $final_app_status  ==  $status ]
then
echo "Parquet data inserted successfully using spark code"
else
echo "Spark Job Failed.Please check the log"
exit 1
fi

#Refresh table
beeline -u $BEELINE_CMD  --outputformat=tsv2 --showHeader=false -e "msck repair table ${cdl_schema_name_wh}.${parquet_table_name}";
#Check Script successfully executed or not
if [ $? -eq 0 ]; then
    echo "Succefully refreshed the table ${cdl_schema_name_wh}.${parquet_table_name} "
else

echo "Beeline command to refresh ${cdl_schema_name_wh}.${parquet_table_name} table failed"
exit 1
fi


#Checking script execution success/failure
insrtRC=$?
fnStatusCheck $insrtRC "echo insert to parquet from avro completed .... $cdl_schema_name_wh is Loaded"  "Error Loading  loading $cdl_table_name"

fnLogMsg INFO "CDL load process is completed for table : $cdl_table_names"



#Load  start end Timestamp
rm $CDL_LOGS/ParquetRun.txt
LoadEndDate=`date +%Y:%m:%d:%H:%M:%S`
echo $LoadEndDate >>$CDL_LOGS/ParquetRun.txt
echo "Avro Parquet Insert Job Completed"
echo $LoadEndDate


echo "Writing into param file for Audit log"
echo $1
echo export CDH_LOAD_LOG_KEY=$CDH_LOAD_LOG_KEY >>$1
echo export LoadStartDate=$LoadStartDate >>$1
echo export LoadEndDate=$LoadEndDate >>$1

else

echo "ParquetTable:Data Already inserted.Please Check Parquet Table Data"

fi

DL_BTEQ_WRAPPER_SRIPT

#!/bin/bash
#====================================================================================================================================================
# Title            : CDL BTEQ Wrapper Script to Load Stage Tables
# ProjectName      : CDL
# Filename         : CDL_BTEQ_WRAPPER.sh
# Description      : Script to load CDL stage tables on Teradata
# Developer        : xxxxxx
# Created on       : JUNE 2017
# Location         : xxxxxx
# Logic            :
# Parameters       : Parameter file name
#Execution          :sh CDL_BTEQ_Wrapper.sh "parameter file name"
# Return codes     :
# Date                         Ver#     Modified By(Name)                 Change and Reason for Change
# ----------    -----        -----------------------------               --------------------------------------
# 2017/06/13                      1     Initial Version

#  ***************************************************************************************************************************

######### This script does the following
######### 1. Create a record in the teradata CDL_LOAD_LOG
######### 2. Execute the BTEQ DML to insert data into stage table - Only if needed
######### 3. Excecute script to get the teradata source table count for Reconciliation
echo "In Bteq script"
echo $2
echo $0
BIN_DIR=`dirname $0`
echo "BIN DIR name"
echo $BIN_DIR
echo source $BIN_DIR/cdl-bl-db-env.conf $2
 source $BIN_DIR/cdl-bl-db-env.conf $2
echo "DB Name"
echo $DB
#Source Hadoop Conf file
source $BIN_DIR/cdl-bl-env.conf

#Source the parameter file
echo $1
source $1

echo  $CDL_CONF
echo  $CDL_PARAM
echo  $CDL_LOGS

# Moving Source env variables to Hadoop_param.conf


#Job Unix Log File Location
LogPath=/home/$USER
RunDate=`date +%Y:%m:%d:%H:%M:%S`
SummaryLogFileNm=CDL_BTEQ_RUN_`date +%Y%m%d%H%M%S`_${TBL_NM}_$USER.log
SLogFullFileName=$CDL_LOGS/${SummaryLogFileNm}
exec 1> $SLogFullFileName 2>&1



#Source bash_profile to get credentials for Teradata BTEQ Script
source ~/.bash_profile
echo "Logon server name"
echo $LOGON

#Remove the done files
echo "*******************Removing Done files**********************"
#echo "Location"
#echo ${DONE_LOG}/donefiles

#listfile=`echo "$1" | awk -F / '{print $NF}'`
#echo "conf file name***************************"
#echo $listfile

#conf_file=${listfile}
#lst_file='.lst'
#listfile_name="${conf_file/.conf/$lst_file}"

#echo "result string"
#echo $listfile_name
echo "list file name***************************"
echo $listfile_name


if [  -f ${DONE_LOG}/donefiles/CDL_AUDIT_Wrapper.sh.${listfile_name}.4.done ]
then
  echo "Audit Done File exist."
  rm ${DONE_LOG}/donefiles/CDL_AUDIT_Wrapper.sh.${listfile_name}.4.done
  echo "Audit Done File CDL_AUDIT_Wrapper.sh.${listfile_name}.4.done deleted"
 if [ $? -gt 0 ]; then
        echo "Delete CDL_AUDIT_Wrapper.sh.${listfile_name}.4.done failed "
        #exit 1
 fi

 else
echo "CDL_AUDIT_Wrapper.sh.${listfile_name}.4.done file not exist"
fi


################################### STEP -1 : Inserting a record into TERADATA Audit Log table  - CDL_LOAD_LOG

echo "STEP -1 : Inserting a record into TERADATA Audit Log table  - CDL_LOAD_LOG"

sh $CDL_BIN/BTEQ/BTEQ_CDL_LOAD_LOG.sh $cdl_table_name $SUBJ_AREA_NM $PROCESS_NM CURRENT_TIMESTAMP $LOAD_END_DTM $PUB_IND

#Check Script successfully executed or not
if [ $? -eq 0 ]; then
    echo "Succefully Inserted record in Teradata CDL_LOAD_LOG table "
else
    echo "Teradata load log entry failed"
    exit 1
fi

#Echo Variables
BTEQ_LOAD_START_DT=$BTEQ_LOAD_START_DATE
BTEQ_LOAD_END_DT=$BTEQ_LOAD_END_DATE

echo "Start End  date for BTEQ"
echo $BTEQ_LOAD_START_DATE
echo  $BTEQ_LOAD_END_DATE


################################### STEP -2 : Insert Data Into Stage table On Teradata ( If Needed )
echo "STEP -2 : Insert Data Into Stage table On Teradata ( If Needed )"

#Echo Variables
STAGE_TABLE_NEEDED=$STAGE_TABLE_NEEDED
BTEQ_LOAD_START_DT=$BTEQ_LOAD_START_DATE
BTEQ_LOAD_END_DT=$BTEQ_LOAD_END_DATE
echo "Start End  date for BTEQ"
echo $STAGE_TABLE_NEEDED
echo $BTEQ_LOAD_START_DATE
echo $BTEQ_LOAD_END_DATE
echo $TBL_NM
echo $DB
echo $SOURCE_CON

#Check whether BTEQ Script execution to INSERT DATA INTO STAGE TABLE is required or not

if [ "$STAGE_TABLE_NEEDED" == "YES" ] ; then

echo "Run BTEQ to load data into Teradata stage table in CDL_ETL_TEMP"

sh $CDL_BIN/BTEQ/BTEQ_${TBL_NM}_LOAD.sh $TBL_NM $TDCH_DB "$BTEQ_LOAD_START_DT" $BTEQ_LOAD_END_DT "${SOURCE_CON}" $BTEQ_DATABASE_NM

#Check Script successfully executed or not
if [ $? -eq 0 ]; then
    echo "Succefully Executed BTEQ LOAD Script"
else
    echo "BTEQ LOAD Script failed"
    exit 1
fi

else
echo "BTEQ Execution is not required"

fi

#### Deleting the existing recon file ####
if [ -f $CDL_LOGS/Reconciliation/Teradata_count_recon_$TBL_NM.txt ]; then

    rm $CDL_LOGS/Reconciliation/Teradata_count_recon_$TBL_NM.txt
    echo "$?"
    if [ $? -gt 0 ]; then
    echo "Deleting existing recon file failed."
    exit 1
    fi

fi

################################### STEP -3 : Get The Teradata Source Table Count
echo "STEP -3 : Teradata Source Table Count"

sh $CDL_BIN/BTEQ/BTEQ_TERADATA_RECON_COUNT.sh $TBL_NM $TDCH_DB "${SOURCE_CON}"

#Check Script successfully executed or not
if [ $? -eq 0 ]; then
    echo "Succefully Executed RECON Script"
else
    echo "RECON Script failed"
    exit 1
fi

DL_TDCH_SCRIPT

#!/usr/bin/bsh
#====================================================================================================================================================
# Title            : DL_TDCH_WRAPPER
# ProjectName      : CDL BaseLine Loads
# Filename         : CDL_TDCH_WRAPPER.sh
# Description      : Script to execute TDCH to extract source data and load into Hadoop in Avro file format.
# Developer        : xxxxxx
# Created on       : JUNE 2017
# Location         : xxxxxxx
# Logic            :
# Parameters       : Parameter file name
# Execution        :sh DL_TDCH_Wrapper.sh "parameterfile name
# Return codes     :
# Date                         Ver#     Modified By(Name)                 Change and Reason for Change
# ----------    -----        -----------------------------               --------------------------------------
# 2017/06/13                      1     Initial Version

#  ***************************************************************************************************************************

######### This script does the following
######### 1. Execute TDCH  to import data in Avro format
######### 2. Write LoadstartDate And LoadEndDate in param file

#Source DBSRVR config file
#Taking user input and passing to DBSRVR file

echo $0
BIN_DIR=`dirname $0`
echo $BIN_DIR
source $BIN_DIR/cdl-bl-db-env.conf $2
#Source Hadoop Conf file
source $BIN_DIR/cdl-bl-env.conf

####Sourcing kerebros ticket to pick up latest one###
source $HOME/.bash_profile
echo $KRB5CCNAME

echo $TDCHQUERYBAND
echo $TERADATA_TDCH_URL
#Source Hadoop Conf file
#source $BIN_DIR/HADOOP_PARM.conf
#. /dv/app/ve2/cdp/edwd/phi/no_gbd/r000/bin/HADOOP_PARM.conf

#Source the parameter file
echo $1
source $1
#echo $2
#echo " THIS IS THE SECOND ARGUEMENT"

echo  $CDL_CONF
echo  $CDL_PARAM
echo  $CDL_LOGS


#Source bash_profile to get credentials
#source ~/.bash_profile

#Job Unix Log File Location
JobRunDate=`date +%Y:%m:%d:%H:%M:%S`
SummaryLogFileNm=CDL_TDCH_`date +%Y%m%d%H%M%S`_${TBL_NM}_$USER.log
SLogFullFileName=$CDL_LOGS/${SummaryLogFileNm}
exec 1> $SLogFullFileName 2>&1


error_count=0

fnLogMsg()
{
     LogTime=`date +%Y%m%d%H%M%S`
     LogMsg="$LogTime:$1:$2"
     echo $LogMsg >> $SLogFullFileName
     echo $LogMsg
}

fnStatusCheck()
{
     CommandRC="$1"
     SuccessMSG="$2"
     FailureMSG="$3"
             
    if [ $CommandRC -eq 0 ]
    then
        fnLogMsg INFO "${SuccessMSG}"
        else
            fnLogMsg INFO "${FailureMSG}"
        exit 1
        error_count=`expr $error_count + 1`
    fi
}




# Variables
teradata_table_name=$TBL_NM
teradata_table_name_for_hive="'$teradata_table_name'"
echo "Hive query table name"
echo $teradata_table_name_for_hive

echo "STEP -1 : Execute the TDCH job to import the data into HDFS in Avro Format"

#Echo variables
echo $TBL_NM
echo $HADOOP_PATH
echo $SOURCE_CON
echo $NUM_MAPPERS
echo $SOURCE_TARGET_FIELD_NAMES
echo $AVRO_SCHEMA_FILE_NAME
echo $TERADATA_TDCH_URL
echo "TDCH Seperator"
echo $TDCH_Seperator


#Job End Date
JobRunDate=`date +%Y:%m:%d:%H:%M:%S`
echo $JobRunDate



#Create directory if not exist
tdch_avro_schema=$CDL_DDL/Avro_Schema
if [[ ! -e $tdch_avro_schema ]]; then
    mkdir $tdch_avro_schema
elif [[  -d $tdch_avro_schema ]]; then
    echo "$tdch_avro_schem already exist"
fi

# copying schema to local path

AvSchemaFile=$tdch_avro_schema/$AVRO_SCHEMA_FILE_NAME

if [ -f "$AvSchemaFile" ]
then
    rm $AvSchemaFile

   echo "AvSchemaFile Temp  schema File deleted"
fi

AvroSchema=$tdch_avro_schema/$AVRO_SCHEMA_FILE_NAME

echo "Copying Avro Schema Filei from HDFS to APP folder"

hadoop fs -copyToLocal $AVRO_TEMP_LOCATION/$AVRO_SCHEMA_FILE_NAME $tdch_avro_schema/

if [ $? -gt 0 ]; then
echo "Avro schema copy to local folder failed"
exit 1
fi

#echo "Avro Location"
#echo hadoop fs -copyToLocal $AVRO_TEMP_LOCATION/$AVRO_SCHEMA_FILE_NAME $tdch_avro_schema/


hadoop fs -test -d $HADOOP_PATH

if [ $? -eq 0 ] ; then
echo "Avro data files exists. It will be deleted"
hadoop fs -rmr -skipTrash $HADOOP_PATH
else
echo " No older avro data files exists"
fi

echo $USER
echo TDCHMETHODS : $NUM_MAPPERS

###################Defaulting mapnum to 19 in test regions due to restriction on repliation factor #########
if [ hostname="dwbdtest1r1e.wellpoint.com" ]
then
  #NUM_MAPPERS="19"
  QUEUE_NAME="cdl_yarn"
else
 QUEUE_NAME="cdl_yarn"
fi

#Remove Temp Files if exist
TeradataReconFile=$CDL_LOGS/Reconciliation/TeradataRecon_count_recon_${TBL_NM}_$RUN_NB.txt
echo "TeradataReconFile"
echo $TeradataReconFile

if [ -f "$TeradataReconFile" ]
then
    rm $TeradataReconFile

   echo "TeradataReconFile Temp File deleted"
fi

#Get the recon count
cat $CDL_LOGS/CDL_BTEQ_RECON_${TBL_NM}_$USER.log|grep 'Count' -A 2|tail -1 >> $CDL_LOGS/Reconciliation/TeradataRecon_count_recon_${TBL_NM}_$RUN_NB.txt


#Getting the count  records
TeradataRecon_count=`cat $CDL_LOGS/Reconciliation/TeradataRecon_count_recon_${TBL_NM}_$RUN_NB.txt`





echo " Teradata COUNT***********"
echo $TeradataRecon_count

echo "Removing spaces"
TeradataRecon_count=`echo ${TeradataRecon_count//[[:blank:]]/}`


#exporting count for Audit Wrapper
echo export TeradataRecon_count=$TeradataRecon_count >>$1

if [ "$TeradataRecon_count" != "0" ];
then

echo "Avro Schema File"
echo file:////$tdch_avro_schema/$AVRO_SCHEMA_FILE_NAME
echo    hadoop jar $TDCH_JAR com.teradata.connector.common.tool.ConnectorImportTool \
-Dmapreduce.job.queuename=$QUEUE_NAME \
        -libjars $HIVE_LIB_JARS -classname com.teradata.jdbc.TeraDriver \
        -url $TERADATA_TDCH_URL \
        -username $TD_DB_ACCT -password $PASSWORD \
        -queryband $TDCHQUERYBAND \
        -jobtype hdfs \
        -fileformat avrofile  \
        -sourcetable $teradata_table_name \
        -nummappers $NUM_MAPPERS \
        -method $TDCHMETHODS \
        -separator $TDCH_Seperator \
        -targetpaths $HADOOP_PATH\
        -avroschemafile file:////$tdch_avro_schema/$AVRO_SCHEMA_FILE_NAME \
        -sourceconditions "$SOURCE_CON"

echo "Executing TDCH........."
#TDCH import command
hadoop jar $TDCH_JAR com.teradata.connector.common.tool.ConnectorImportTool \
-Dmapreduce.job.queuename=$QUEUE_NAME \
-libjars $HIVE_LIB_JARS -classname com.teradata.jdbc.TeraDriver \
-url $TERADATA_TDCH_URL \
-username $TD_DB_ACCT -password $PASSWORD \
        -queryband $TDCHQUERYBAND \
-jobtype hdfs \
-fileformat avrofile  \
-sourcetable $teradata_table_name \
-nummappers $NUM_MAPPERS \
        -method $TDCHMETHODS \
-separator $TDCH_Seperator \
-targetpaths $HADOOP_PATH\
-avroschemafile file:////$tdch_avro_schema/$AVRO_SCHEMA_FILE_NAME \
-sourceconditions "$SOURCE_CON"


insrtRC=$?
fnStatusCheck $insrtRC "$teradata_table_name is Loaded"  "Error in loading $teradata_table_name"

fnLogMsg INFO "CDL load process into Avro format is complete for the table  : $teradata_table_name, JobRunDate : $JobRunDate"

echo "Avro Parquet Insert Job Completed"

else
echo " Teradata records are Zero"
hdfs dfs -mkdir $HADOOP_PATH
echo hdfs dfs -mkdir $HADOOP_PATH
fi
#Job End Date
LoadEndDate=`date +%Y:%m:%d:%H:%M:%S`
echo $LoadEndDate
JobEndTime=$LoadEndDate

echo "Job end Time TDCH"
echo $JobEndTime


#Exporting  and appending variables  to Param file for Audit script

echo "Writing variables in file"
echo /home/$USER/$1
echo export JobRunDate=$JobRunDate >>$1
echo export JobEndTime=$JobEndTime>>$1

Apache Spark : RDD vs DataFrame vs Dataset

With Spark2.0 release, there are 3 types of data abstractions which Spark officially provides now to use : RDD,DataFrame and DataSet .

For a new user, it might be confusing to understand relevance of each one and decide which one to use and which one not to. In this post, will discuss each one of them in detail with their differences and pros-cons.

Short Combined Intro :

Before i discuss each one in detail separately, want to start with a short combined intro.

Evolution of these abstractions happened in this way :

RDD (Spark1.0) —> Dataframe(Spark1.3) —> Dataset(Spark1.6)

RDD being the oldest available from 1.0 version to Dataset being the newest available from 1.6 version.

Given same data, each of the 3 abstraction will compute and give same results to user. But they differ in performance and the ways they compute.

RDD lets us decide HOW we want to do which limits the optimisation Spark can do on processing underneath where as dataframe/dataset lets us decide WHAT we want to do and leave everything on Spark to decide how to do computation. 

We will understand this in 2 minutes what is meant by HOW & WHAT .

Dataframe came as a major performance improvement over RDD but not without some downsides.

This led to development of Dataset which is an effort  to unify best of RDD and data frame.

In future, Dataset will eventually replace RDD and Dataframe to become the only API spark users should be using in code.

Lets understand them in detail one by one.

RDD:

-Its building block of spark. No matter which abstraction Dataframe or Dataset we use, internally final computation is done on RDDs. 

-RDD is lazily evaluated immutable parallel collection of objects exposed with lambda functions.

-The best part about RDD is that it is simple. It provides familiar OOPs style APIs with compile time safety. We can load any data from a source,convert them into RDD and store in memory to compute results. RDD can be easily cached if same set of data needs to recomputed.

-But the disadvantage is performance limitations. Being in-memory jvm objects, RDDs involve overhead of Garbage Collection and Java(or little better Kryo) Serialisation which are expensive when data grows.

RDD example:

Dataframe:

-DataFrame is an abstraction which gives a schema view of data.  Which means it gives us a view of data as columns with column name and types info, We can think data in data frame like a table in database.

-Like RDD, execution in Dataframe too is lazy triggered .

-offers huge performance improvement over RDDs because of 2 powerful features it has: 

1. Custom Memory management  (aka Project Tungsten)

Data is stored in off-heap memory in binary format. This saves a lot of memory space. Also there is no Garbage Collection overhead involved. By knowing the schema of data in advance and storing efficiently in binary format, expensive java Serialization is also avoided.

2. Optimized Execution Plans        (aka Catalyst Optimizer)

       Query plans are created for execution using Spark catalyst optimiser. After an optimised execution plan is prepared going through some steps, the final execution happens internally on RDDs only but thats completely hidden from the users.

Execution plan stages

Just to give an example of optimisation with respect to the above picture, lets consider a query as below  :

inefficient query: filter after join

In the above query, filter is used before join which is a costly shuffle operation. The logical plan sees that and in optimised logical plan, this filter is pushed to execute before join. In the optimised execution plan, it can leverage datasource capabilities also and push that filter further down to datasource so that it can apply that filter on the disk level rather than bringing all data in memory and doing filter in memory (which is not possible while directly using RDDs). So filter method now effectively works like a WHERE clause in a database query. Also with optimised data sources like parquet , if Spark sees that you need only few columns to compute the results , it will read and fetch only those columns from parquet saving both disk IO and memory.

-Drawback :

 Lack of Type Safety. As a developer, i will not like using dataframe as it doesn't seem developer friendly. Referring attribute by String names means no compile time safety. Things can fail at runtime. Also APIs doesn't look programmatic and more of sql kind.

Dataframe example:  

 2 ways to define: 

  

 1. Expression BuilderStyle   

 2. SQL Style

As discussed, If we try using some columns not present in schema, we will get problem only at runtime . For example, if we try accessing salary when only name and age are present in the schema will exception like below:

Dataset:

-It is an extension to Dataframe API, the latest abstraction which tries to provide best of both RDD and Dataframe.

-comes with OOPs style and developer friendly compile time safety like RDD as well as performance boosting features of Dataframe : Catalyst optimiser and custom memory management.

-How dataset scores over Dataframe is an additional feature it has: Encoders .

-Encoders act as interface between JVM objects and off-heap custom memory binary format data. 

-Encoders generate byte code to interact with off-heap data and provide on-demand access to individual attributes without having to de-serialize an entire object.

-case class is used to define the structure of data schema in Dataset. Using case class, its very easy to work with dataset. Names of different attributes in case class is directly mapped to attributes in Dataset . It gives feeling like working with RDD but actually underneath it works same as Dataframe.

Dataframe is infact treated as dataset of generic row objects. DataFrame=Dataset[Row] . So we can always convert a data frame at any point of time into a dataset by calling ‘as’ method on Dataframe.
 e.g.  df.as[MyClass]

Dataset Example :

Important point  to remember is that both Dataset and DataFrame internally does final execution on RDD objects only but the difference is users do not write code to create the RDD collections and have no control as such over RDDs.  RDDs are created in the execution plan as last stage after deciding and going through all the optimizations (see Execution Plan Diagram). 

Thats why at the beginning of this post i emphasized on……..RDD let us decide HOW we want to do where as Dataframe/Dataset lets us decide WHAT we want to do.  

And all these optimisations could have been possible because data is structured and Spark knows about the schema of data in advance. So it can apply all the powerful features like tungsten custom memory off-heap binary storage,catalyst optimiser and encoders to get the performance which was not possible if users would have been directly working on RDD.

Conclusion:

In short, Spark is moving from unstructured computation(RDDs) towards structured computation because of many performance optimisations it allows . Data frame was a step in direction of structured computation but lacked developer friendliness of compile time safety,lambda functions. Finally Dataset is the unification of Dataframe and RDD to bring the best abstraction out of two.

Going forward developers should only be concerned about DataSet while Dataframe and RDD will be discouraged to use. But its always better to be aware of the legacy for better understanding of internals.

Interestingly,most of these new concepts like custom memory management(tungsten),logical/physical plans(catalyst optimizer),encoders(dataset),etc seems to be inspired from its competitor Apache Flink which inherently supports these since inception.There are other new powerful feature enhancements like windowing,sessions,etc coming in Spark which are already in Flink. So its better to keep a close watch on both Spark and Flink in coming days.