Pyspark Use case 2

Pyspark Use case 2

---------

how to upgrade spark version

pip install pyspark==2.3.2

from pyspark.sql import SparkSession

spark=SparkSession.builder.master('local').appName('sparkapp').getOrCreate()

spark

Sample.txt
----------------

spark spark spark spark spark spark spark

spark spark spark spark spark

spark

spark spark spark

r1=spark.sparkContext.textFile("c:\\Users\\Nethra\\Desktop\\sample.txt")

print(r1.collect());

[u'spark spark spark spark spark spark spark', u'spark spark spark spark spark ', u'spark', u'spark spark spark']

print(r1.count())

4

r2=r1.map(lambda x:(str)(x)).map(lambda x:x.split(" "))

print(r2.collect())

[['spark spark spark spark spark spark spark'], ['spark spark spark spark spark '], ['spark'], ['spark spark spark']]

Get the number of columns:

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

r3=r2.map(lambda x:len(x))

print(r3.collect())

[7, 6, 1, 3]

Maximum column in the list:

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

columnsize=max(r3.collect())

print(columnsize)

7

#a function to fill the words missing with  "nodata"

#input is a list of variable size and output will be a list of fixed size(columns)

def fillmissing(x):

    result= []

    for i in range(columnsize):

        try:

            result.append(x[i])

        except:

            result.append('nodata')

    return result

print(fillmissing(['spark','spark']))

       
resultdata=r2.map(lambda x:fillmissing(x))

print(resultdata.collect())

[['spark', 'spark', 'spark', 'spark'], ['spark', 'spark', 'spark', 'spark'], ['spark', 'nodata', 'nodata', 'nodata'], ['spark', 'spark', 'spark', 'nodata']]

# Convert RDD into dataframe

df1=resultdata.toDF(['c1','c2','c3','c4'])

df1.show()

+-----+------+------+------+

|   c1|    c2|    c3|    c4|

+-----+------+------+------+

|spark| spark| spark| spark|

|spark| spark| spark| spark|

|spark|nodata|nodata|nodata|

|spark| spark| spark|nodata|

+-----+------+------+------+

pip install pyspark==2.3.2

df2=spark.read.format('csv').option('inferSchema',True).option('header',False).load("file:///home/hadoop/sample1.csv")

df2.show(5)

+---+---+-------------------+

|_c0|_c1|                _c2|

+---+---+-------------------+

|  1|hyd|2013-12-24 23:14:45|

|  2|chn|2018-11-30 11:12:34|

from pyspark.sql.functions import sqrt

from pyspark.sql.types import IntegerType

df.select('Age',sqrt('Age')).show()

>> df2.select('_c0',sqrt('_c0')).show()

+---+------------------+

|_c0|         SQRT(_c0)|

+---+------------------+

|  1|               1.0|

|  2|1.4142135623730951|

#create a UDF

def f2(x):

   return (x*x-x)/2

   print(f2(20))

df3=df2.na.drop()

+---+---+-------------------+

|_c0|_c1|                _c2|

+---+---+-------------------+

|  1|hyd|2013-12-24 23:14:45|

|  2|chn|2018-11-30 11:12:34|

df3=df2.dropna()

+---+---+-------------------+

|_c0|_c1|                _c2|

+---+---+-------------------+

|  1|hyd|2013-12-24 23:14:45|

|  2|chn|2018-11-30 11:12:34|

#Registering the UDF with SparkSQL function

spark.udf.register('myfunc',f2,IntegerType())

myfunc=udf(f2,IntegerType())

df3.select('Age',myfunc('Age')).show()

from pyspark.sql import Row

r1 =Row(age=20,salary=20000,name=None,date_time='2014-12-23')

r2=Row(salary=40000,name='shanker',age=None,date_time=None)

r3=Row(age=30,name='priya',salary=40000,date_time='2015-11-10')

df4=spark.createDataFrame([r1,r2,r3,r4])

df4.printSchema()

+----+----------+-------+------+                                               

| age| date_time|   name|salary|

+----+----------+-------+------+

|  20|2014-12-23|   null| 20000|

|null|      null|shanker| 40000|

|  30|2015-11-10|  priya| 40000|

+----+----------+-------+------+

>>> df4.printSchema()

root

 |-- age: long (nullable = true)

 |-- date_time: string (nullable = true)

 |-- name: string (nullable = true)

 |-- salary: long (nullable = true)

df4.show()

df4.na.fill(-1).na.fill('notprovided').show()

df4.na.fill(-1).fill('notprovided').show()

+---+-----------+-----------+------+

|age|  date_time|       name|salary|

+---+-----------+-----------+------+

| 20| 2014-12-23|notprovided| 20000|

| -1|notprovided|    shanker| 40000|

| 30| 2015-11-10|      priya| 40000|

+---+-----------+-----------+------+

from pyspark.sql.functions import col

>> df5=df4.withColumn('date_time',col("date_time").cast('timestamp'))

>>> df5.printSchema()

root

 |-- age: long (nullable = true)

 |-- date_time: timestamp (nullable = true)

 |-- name: string (nullable = true)

 |-- salary: long (nullable = true)

df5.select('date_time',year('date_time')).show()

------------------+---------------+

|          date_time|year(date_time)|

+-------------------+---------------+

|2014-12-23 00:00:00|           2014|

|               null|           null|

|2015-11-10 00:00:00|           2015|

+-------------------+---------------+

csv file:

1,hyd,2013-12-24 23:14:45

2,chn,2018-11-30 11:12:34

df6=spark.read.format('csv').option('inferSchema',True).load("file:///home/hadoop/sample1.csv"")

df6.printSchema()

from pyspark.sql.functions import max ===> error is thrown on the below statement

from__builtin__ import max  ==> overriding the max function

l1=[12,23,34,45]

print(max[l1])

df.dropDuplicates().show()  ==>using tab

Spark with MongoDB - Read / Write Operations

Spark with MongoDB - Read / Write Operations

$ mongo // start CLI

MongoDB shell version v3.6.3
connecting to: mongodb://127.0.0.1:27017
MongoDB server version: 3.6.3

// Adding 2 records in person
> db.person.insert([{
   "id":100,
           "name":"Sankara",
           "salary":3000,
   "city":"Pallathur"
         
        },
        {
           "id":101,
           "name":"Rasee",
           "salary":3100,
   "city":"Kanadukathan"
        }])
BulkWriteResult({
"writeErrors" : [ ],
"writeConcernErrors" : [ ],
"nInserted" : 2,
"nUpserted" : 0,
"nMatched" : 0,
"nModified" : 0,
"nRemoved" : 0,
"upserted" : [ ]
})

// select
> db.person.find()
{ "_id" : ObjectId("5c780db0d44e3fc2ebd26d43"), "id" : 100, "name" : "Sankara", "salary" : 3000, "city" : "Pallathur" }
{ "_id" : ObjectId("5c780db0d44e3fc2ebd26d44"), "id" : 101, "name" : "Rasee", "salary" : 3100, "city" : "Kanadukathan" }

// beautify
> db.person.find().pretty()
{
"_id" : ObjectId("5c780db0d44e3fc2ebd26d43"),
"id" : 100,
"name" : "Sankara",
"salary" : 3000,
"city" : "Pallathur"
}
{
"_id" : ObjectId("5c780db0d44e3fc2ebd26d44"),
"id" : 101,
"name" : "Rasee",
"salary" : 3100,
"city" : "Kanadukathan"
}

//search with condition
> db.person.find({"name":"Sankara"})
{ "_id" : ObjectId("5c780db0d44e3fc2ebd26d43"), "id" : 100, "name" : "Sankara", "salary" : 3000, "city" : "Pallathur" }

//AND and OR conditions

db.person.find({$and:[{"name":"Sankara"},{"id":101}]})
> db.person.find({$or:[{"name":"Sankara"},{"id":101}]})

{ "_id" : ObjectId("5c799a6092b6d263e5f35ba8"), "id" : 100, "name" : "Sankara", "salary" : 3000, "city" : "Pallathur" }
{ "_id" : ObjectId("5c799a6092b6d263e5f35ba9"), "id" : 101, "name" : "Rasee", "salary" : 3100, "city" : "Kanadukathan" }
>

//How to update the record in mongoDB

db.person.find()

{ "_id" : ObjectId("5c799a6092b6d263e5f35ba8"), "id" : 100, "name" : "Sankara", "salary" : 3000, "city" : "Pallathur" }
{ "_id" : ObjectId("5c799a6092b6d263e5f35ba9"), "id" : 101, "name" : "Rasee", "salary" : 3100, "city" : "Kanadukathan" }
{ "_id" : ObjectId("5c79a09765e80e21e2e727ce"), "id" : 500, "name" : "Sathya", "city" : "Bangalore", "salary" : 3000 }
{ "_id" : ObjectId("5c79a09765e80e21e2e727cf"), "id" : 501, "name" : "Lakshmi", "city" : "Chennai", "salary" : 3100 }
{ "_id" : ObjectId("5c79a09765e80e21e2e727d0"), "id" : 502, "name" : "Kalai", "city" : "Perai", "salary" : 3200 }

> db.person.update({"id":100},{$set:{"name":"Arun"}})
WriteResult({ "nMatched" : 1, "nUpserted" : 0, "nModified" : 1 })

> db.person.find()

{ "_id" : ObjectId("5c799a6092b6d263e5f35ba8"), "id" : 100, "name" : "Arun", "salary" : 3000, "city" : "Pallathur" }
{ "_id" : ObjectId("5c799a6092b6d263e5f35ba9"), "id" : 101, "name" : "Rasee", "salary" : 3100, "city" : "Kanadukathan" }
{ "_id" : ObjectId("5c79a09765e80e21e2e727ce"), "id" : 500, "name" : "Sathya", "city" : "Bangalore", "salary" : 3000 }
{ "_id" : ObjectId("5c79a09765e80e21e2e727cf"), "id" : 501, "name" : "Lakshmi", "city" : "Chennai", "salary" : 3100 }
{ "_id" : ObjectId("5c79a09765e80e21e2e727d0"), "id" : 502, "name" : "Kalai", "city" : "Perai", "salary" : 3200 }

// Spark starts here
spark-shell --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.0
 
import com.mongodb.spark._
import com.mongodb.spark.config.ReadConfig
import com.mongodb.spark.sql._
 import org.bson.Document
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("MongoPerson").master("local[*]").getOrCreate()
val readConfig = ReadConfig(Map("uri" -> "mongodb://127.0.0.1/", "database" -> "test", "collection" -> "person"))
val df = spark.read.mongo(readConfig)  // test.person content populated here


scala> df.show
+--------------------+------------+-----+-------+------+
|                 _id|        city|   id|   name|salary|
+--------------------+------------+-----+-------+------+
|[5c780db0d44e3fc2...|   Pallathur|100.0|Sankara|3000.0|
|[5c780db0d44e3fc2...|Kanadukathan|101.0|  Rasee|3100.0|
+--------------------+------------+-----+-------+------+


scala> df.select("city","id","name","salary").show
+------------+-----+-------+------+
|        city|   id|   name|salary|
+------------+-----+-------+------+
|   Pallathur|100.0|Sankara|3000.0|
|Kanadukathan|101.0|  Rasee|3100.0|
+------------+-----+-------+------+

scala> df.createOrReplaceTempView("tblPerson") // SparkSQL

scala> spark.sql("select * from tblPerson").show
+--------------------+------------+-----+-------+------+
|                 _id|        city|   id|   name|salary|
+--------------------+------------+-----+-------+------+
|[5c780db0d44e3fc2...|   Pallathur|100.0|Sankara|3000.0|
|[5c780db0d44e3fc2...|Kanadukathan|101.0|  Rasee|3100.0|
+--------------------+------------+-----+-------+------+

scala> spark.sql("select id,name,city,salary from tblPerson").filter("salary==3000").show
+-----+-------+---------+------+
|   id|   name|     city|salary|
+-----+-------+---------+------+
|100.0|Sankara|Pallathur|3000.0|
+-----+-------+---------+------+


// Making In-memory collection in Spark
scala> case class Person(id:Int, name:String, city:String, salary:Int)
defined class Person

scala> val ob1 = new Person(500,"Sathya","Bangalore",3000)
ob1: Person = Person(500,Sathya,Bangalore,3000)

scala> val ob2 = new Person(501,"Lakshmi","Chennai",3100)
ob2: Person = Person(501,Lakshmi,Chennai,3100)

scala> val ob3 = new Person(502,"Kalai","Perai",3200)
ob3: Person = Person(502,Kalai,Perai,3200)

scala> val r1 = sc.parallelize(List(ob1,ob2,ob3))
r1: org.apache.spark.rdd.RDD[Person] = ParallelCollectionRDD[40] at parallelize at <console>:40

scala> r1.collect.foreach(println)
Person(500,Sathya,Bangalore,3000)
Person(501,Lakshmi,Chennai,3100)
Person(502,Kalai,Perai,3200)

// Converting inmemory collection into Dataframe
scala> val dfCollection = r1.toDF
dfCollection: org.apache.spark.sql.DataFrame = [id: int, name: string ... 2 more fields]

scala> dfCollection.printSchema
root
 |-- id: integer (nullable = false)
 |-- name: string (nullable = true)
 |-- city: string (nullable = true)
 |-- salary: integer (nullable = false)


scala> dfCollection.show
+---+-------+---------+------+
| id|   name|     city|salary|
+---+-------+---------+------+
|500| Sathya|Bangalore|  3000|
|501|Lakshmi|  Chennai|  3100|
|502|  Kalai|    Perai|  3200|
+---+-------+---------+------+



// Going to write in-memory collection into Mongo
scala> import com.mongodb.spark.config.{ReadConfig, WriteConfig}
import com.mongodb.spark.config.{ReadConfig, WriteConfig}

// Writer Configuration : test.person
scala> val writeConfig = WriteConfig(Map("uri" -> "mongodb://127.0.0.1/test.person"))
writeConfig: com.mongodb.spark.config.WriteConfig.Self = WriteConfig(test,person,Some(mongodb://127.0.0.1/test.person),true,512,15,WriteConcernConfig(None,None,None,None),None,false,true)

scala> MongoSpark.save(dfCollection.write.mode("append"), writeConfig)


// Back to Mongo
db.person.find().pretty()
{
"_id" : ObjectId("5c7815d1d44e3fc2ebd26d47"),
"id" : 100,
"name" : "Sankara",
"salary" : 3000,
"city" : "Pallathur"
}
{
"_id" : ObjectId("5c7815d1d44e3fc2ebd26d48"),
"id" : 101,
"name" : "Rasee",
"salary" : 3100,
"city" : "Kanadukathan"
}
{
"_id" : ObjectId("5c7816f165e80e7efcae207d"),
"id" : 500,
"name" : "Sathya",
"city" : "Bangalore",
"salary" : 3000
}
{
"_id" : ObjectId("5c7816f165e80e7efcae207e"),
"id" : 501,
"name" : "Lakshmi",
"city" : "Chennai",
"salary" : 3100
}
{
"_id" : ObjectId("5c7816f165e80e7efcae207f"),
"id" : 502,
"name" : "Kalai",
"city" : "Perai",
"salary" : 3200
}

// Back to Spark
scala> df.show
+--------------------+------------+---+-------+------+
|                 _id|        city| id|   name|salary|
+--------------------+------------+---+-------+------+
|[5c7815d1d44e3fc2...|   Pallathur|100|Sankara|  3000|
|[5c7815d1d44e3fc2...|Kanadukathan|101|  Rasee|  3100|
|[5c7816f165e80e7e...|   Bangalore|500| Sathya|  3000|
|[5c7816f165e80e7e...|     Chennai|501|Lakshmi|  3100|
|[5c7816f165e80e7e...|       Perai|502|  Kalai|  3200|
+--------------------+------------+---+-------+------+

// Dropping in Mongo
>db.person.drop()
true

// Dataframe is empty
scala> df.show
+---+----+---+----+------+
|_id|city| id|name|salary|
+---+----+---+----+------+
+---+----+---+----+------+

//Create new DB

use Retail

db.retail.person2.insert([{"name":"Sankara","id":101}])

db.retail.person2.find()

Spark with MongoDB integration

Spark with MongoDB integration

<!-- https://mvnrepository.com/artifact/org.mongodb.spark/mongo-spark-connector -->
<dependency>
    <groupId>org.mongodb.spark</groupId>
    <artifactId>mongo-spark-connector_2.11</artifactId>
    <version>2.4.0</version>
</dependency>


spark-shell --packages org.mongodb.spark:mongo-spark-connector_2.11:2.4.0

scala> import com.mongodb.spark._
import com.mongodb.spark._

scala> import com.mongodb.spark.config.ReadConfig
import com.mongodb.spark.config.ReadConfig

scala> import com.mongodb.spark.sql._
import com.mongodb.spark.sql._

scala> import org.bson.Document
import org.bson.Document

scala> import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SparkSession

scala> import org.apache.spark.sql.functions.{max, min}
import org.apache.spark.sql.functions.{max, min}

scala> val spark = SparkSession.builder().appName("MongoPlayers").master("local[*]").getOrCreate()

scala> val readConfig = ReadConfig(Map("uri" -> "mongodb://127.0.0.1/", "database" -> "test", "collection" -> "players"))

scala>  val df = spark.read.mongo(readConfig)

scala>  df.printSchema()
root
 |-- _id: struct (nullable = true)
 |    |-- oid: string (nullable = true)
 |-- age: double (nullable = true)
 |-- birthdate: string (nullable = true)
 |-- birthplace: string (nullable = true)
 |-- height: string (nullable = true)
 |-- id: double (nullable = true)
 |-- imageUrl: string (nullable = true)
 |-- name: string (nullable = true)
 |-- number: double (nullable = true)
 |-- position: string (nullable = true)
 |-- twitterHandle: string (nullable = true)
 |-- twitterURL: string (nullable = true)
 |-- weight: double (nullable = true)


scala> df.show()

Playing with MongoDB

Playing with MongoDB

$ mongo // start CLI

MongoDB shell version v3.6.3

connecting to: mongodb://127.0.0.1:27017
MongoDB server version: 3.6.3

// create a json object
 post = { "position":"Right Wing", "id": 84, "weight":200, "height":"6' 0\"", "imageurl" : "url", "birthplace" : "Seria,BRN", "age":37, "name":"Craig Adams", "birthdate":"April 26,1977", "number":27 }

// insert single
db.players.insert(post);

//insert multiple
db.players.insert([{
         "position":"Right Wing",
         "id":8465166,
         "weight":200,
         "height":"6' 0\"",
         "imageUrl":"http://1.cdn.nhle.com/photos/mugs/8465166.jpg",
         "birthplace":"Seria, BRN",
         "age":37,
         "name":"Craig Adams",
         "birthdate":"April 26, 1977",
         "number":27
      },
      {
         "position":"Right Wing",
         "id":8475761,
         "weight":195,
         "height":"6' 2\"",
         "imageUrl":"http://1.cdn.nhle.com/photos/mugs/8475761.jpg",
         "birthplace":"Gardena, CA, USA",
         "age":23,
         "name":"Beau Bennett",
         "birthdate":"November 27, 1991",
         "number":19
      }])

BulkWriteResult({
"writeErrors" : [ ],
"writeConcernErrors" : [ ],
"nInserted" : 2,
"nUpserted" : 0,
"nMatched" : 0,
"nModified" : 0,
"nRemoved" : 0,
"upserted" : [ ]
})

// select all

db.players.find()

{ "_id" : ObjectId("5c77b1dbd44e3fc2ebd26d2a"), "position" : "Right Wing", "id" : 8465166, "weight" : 200, "height" : "6' 0\"", "imageUrl" : "http://1.cdn.nhle.com/photos/mugs/8465166.jpg", "birthplace" : "Seria, BRN", "age" : 37, "name" : "Craig Adams", "birthdate" : "April 26, 1977", "number" : 27 }
{ "_id" : ObjectId("5c77b1dbd44e3fc2ebd26d2b"), "position" : "Right Wing", "id" : 8475761, "weight" : 195, "height" : "6' 2\"", "imageUrl" : "http://1.cdn.nhle.com/photos/mugs/8475761.jpg", "birthplace" : "Gardena, CA, USA", "age" : 23, "name" : "Beau Bennett", "birthdate" : "November 27, 1991", "number" : 19 }
{ "_id" : ObjectId("5c77b1dbd44e3fc2ebd26d2c"), "position" : "Left Wing", "id" : 8471260, "weight" : 202, "height" : "6' 1\"", "imageUrl" : "http://3.cdn.nhle.com/photos/mugs/8471260.jpg", "birthplace" : "Meadow Lake, SK, CAN", "age" : 29, "name" : "Blake Comeau", "birthdate" : "February 18, 1986", "number" : 17 }

// select with condition
>  db.players.find(
... {"position":"Goalie"}

... )

{ "_id" : ObjectId("5c77b1dbd44e3fc2ebd26d41"), "position" : "Goalie", "id" : 8470594, "weight" : 180, "height" : "6' 2\"", "imageUrl" : "http://3.cdn.nhle.com/photos/mugs/8470594.jpg", "birthplace" : "Sorel, QC, CAN", "age" : 30, "name" : "Marc-Andre Fleury", "birthdate" : "November 28, 1984", "number" : 29 }
{ "_id" : ObjectId("5c77b1dbd44e3fc2ebd26d42"), "position" : "Goalie", "id" : 8471306, "weight" : 220, "height" : "6' 1\"", "imageUrl" : "http://1.cdn.nhle.com/photos/mugs/8471306.jpg", "birthplace" : "Fussen, DEU", "age" : 29, "name" : "Thomas Greiss", "birthdate" : "January 29, 1986", "number" : 1 }


// display the db
> db
test

// display it in pretty way

>  db.players.find().pretty()

{
"_id" : ObjectId("5c77b1dbd44e3fc2ebd26d2a"),
"position" : "Right Wing",
"id" : 8465166,
"weight" : 200,
"height" : "6' 0\"",
"imageUrl" : "http://1.cdn.nhle.com/photos/mugs/8465166.jpg",
"birthplace" : "Seria, BRN",
"age" : 37,
"name" : "Craig Adams",
"birthdate" : "April 26, 1977",
"number" : 27
}
{
"_id" : ObjectId("5c77b1dbd44e3fc2ebd26d2b"),
"position" : "Right Wing",
"id" : 8475761,
"weight" : 195,
"height" : "6' 2\"",
"imageUrl" : "http://1.cdn.nhle.com/photos/mugs/8475761.jpg",
"birthplace" : "Gardena, CA, USA",
"age" : 23,
"name" : "Beau Bennett",
"birthdate" : "November 27, 1991",
"number" : 19
}

// show all tables

>  show collections
players



>  db.players.findOne();
{
"_id" : ObjectId("5c77b1dbd44e3fc2ebd26d2a"),
"position" : "Right Wing",
"id" : 8465166,
"weight" : 200,
"height" : "6' 0\"",
"imageUrl" : "http://1.cdn.nhle.com/photos/mugs/8465166.jpg",
"birthplace" : "Seria, BRN",
"age" : 37,
"name" : "Craig Adams",
"birthdate" : "April 26, 1977",
"number" : 27
}

// remove a record
> db.players.remove({"id":8465166 })
WriteResult({ "nRemoved" : 1 })

// drop a document

> db.players.drop()
true

MongoDB Installation steps in Ubuntu 18.10

MongoDB Installation steps in Ubuntu 18.10

MongoDB Installation:
-----------------------
$ sudo apt update

$ sudo apt install mongodb

$ sudo systemctl status mongodb

$ sudo systemctl start / stop / restart / disable / enable mongodb

$ sudo systemctl start mongodb // start it

$ mongo  // to start with CLI

MongoDB shell version v3.6.3
connecting to: mongodb://127.0.0.1:27017
MongoDB server version: 3.6.3

 post = { "position":"Right Wing", "id": 84, "weight":200, "height":"6' 0\"", "imageurl" : "url", "birthplace" : "Seria,BRN", "age":37, "name":"Craig Adams", "birthdate":"April 26,1977", "number":27 }

Pyspark Hive Integeration

Hive and Spark Integeration:
hive-->warehouse
hive-->data(warehouse)
     -->Metastore(RDBMS)

Hive--->HQL/SQL --> which is used to play the data against HDFS.

metastore--> a dtabase in RDBMS
---------------------------------------------------------------
Oracle-->SQL

data-->files,NTFS,ext4

transactional

metadata-->RDBMS

hive>show tables;

show databases;

show tables in default;

select * from default.accountdevice limit 5
pyspark


spark and hive
----------------
From hadoop conff(hadoophome/etc/hadoop)..core-site.xml and hdfs-site.xml
Hive conf.(hivehome/conf).hive-site.xml

sparkhome/conf->core-site.xml,hdfs-site.xml and hive-site.xml

spark.sql('show databases').count()

spark.sql('select * from default.accountservice limit 10').show(10)

df_hive1=spark.read.table('default.accountdevice')

park.sql('insert into retail.employee values(103,"Sriram")')
DataFrame[]

>>> spark.sql('select * from retail.employee')
DataFrame[id: int, name: string]

>>> spark.sql('select * from retail.employee').show()
+---+-------+
| id|   name|
+---+-------+
|102| RAMESH|
|103| SRiram|
|101|shankar|
|103| Sriram|
+---+-------+

>>> spark.read.table('retail.employee')
DataFrame[id: int, name: string]

>>> spark.read.table('retail.employee').show()
+---+-------+
| id|   name|
+---+-------+
|102| RAMESH|
|103| SRiram|
|101|shankar|
|103| Sriram|
+---+-------+

df_hive1.printSchema()
root
 |-- id: integer (nullable = true)
 |-- name: string (nullable = true)

df_hive1.show(5)
---+-------+
| id|   name|
+---+-------+
|102| RAMESH|
|103| SRiram|
|101|shankar|
|103| Sriram|
+---+-------+
df_hive2=spark.sql('select * from default.accountdevice')

df_hive2.count()

df_hive1.count()

DSL -->domain specific language

df_hive1=spark.read.table('default.accountdevice').filter('device_id=29')

df_hive1.show(5)

df3=df_hive1.withColumn('new',current_timestamp())

df3=df_hive1.withColumn('new',current_date())
+---+-------+------------+
| id|   name|Current_Date|
+---+-------+------------+
|102| RAMESH|  2019-03-02|
|103| SRiram|  2019-03-02|
|101|shankar|  2019-03-02|
|103| Sriram|  2019-03-02|
+---+-------+------------+

from pyspark.sql.functions import *

df3.show(5)

df3.write.saveAsTable('default.testspark')

show tables in default;

select * from testsparkdf limit 10;

spark.sql("select * from testsparkdf2")
DataFrame[id: int, name: string, Current_Date: date]

>>> spark.sql("select * from testsparkdf2").show()
+---+-------+------------+
| id|   name|Current_Date|
+---+-------+------------+
|101|shankar|  2019-03-02|
|102| RAMESH|  2019-03-02|
|103| SRiram|  2019-03-02|
|103| Sriram|  2019-03-02|
+---+-------+------------+


spark is default as parquet format

df3.write.format('csv').saveAsTable('default.testsparkdf2')

df4=spark.read.format('csv'),load("hdfs://172.16.38.131:8020/user/hive/warehouse/
testspark2/,...../csv")

hadoop@hadoop:~$ hadoop fs -ls /user/hive/warehouse/testsparkdf3
Found 5 items
-rw-r--r--   3 hadoop supergroup          0 2019-03-02 04:14 /user/hive/warehouse/testsparkdf3/_SUCCESS
-rw-r--r--   3 hadoop supergroup         22 2019-03-02 04:14 /user/hive/warehouse/testsparkdf3/part-00000-f2e9991d-fbaa-40a7-b2c7-730d61b2898c-c000.csv
-rw-r--r--   3 hadoop supergroup         22 2019-03-02 04:14 /user/hive/warehouse/testsparkdf3/part-00001-f2e9991d-fbaa-40a7-b2c7-730d61b2898c-c000.csv
-rw-r--r--   3 hadoop supergroup         23 2019-03-02 04:14 /user/hive/warehouse/testsparkdf3/part-00002-f2e9991d-fbaa-40a7-b2c7-730d61b2898c-c000.csv
-rw-r--r--   3 hadoop supergroup         22 2019-03-02 04:14 /user/hive/warehouse/testsparkdf3/part-00003-f2e9991d-fbaa-40a7-b2c7-730d61b2898c-c000.csv
hadoop@hadoop:~$

df4.show

df4=spark.read.load("hdfs://172.16.38.131:8020/user/hive/warehouse/
testspark2/,...../csv")

df3.show()-->error is thrown

>>> df4=spark.read.format("csv").load("hdfs://127.0.0.1:9000/user/hive/warehouse/testsparkdf3/")

>>> df4.show()

+---+-------+----------+
|_c0|    _c1|       _c2|
+---+-------+----------+
|101|shankar|2019-03-02|
|102| RAMESH|2019-03-02|
|103| SRiram|2019-03-02|
|103| Sriram|2019-03-02|
+---+-------+----------+

HDFS-->hive(HQL)(MR Execution Engine)
spark--->(spark engine)
with hive-->spark integeration -->we are giving spark the access to HiveData as
well as Schema

spark can read and write from HDFS--(No tabular fashion)

dfs=spark.read.format('csv').option('delimeter','\t').load('file://home...tsv')
df5.show(4)

df5.write.saveAsTable('default.pysparkdf')
External table-->LOCATION

mysql
select * from TBLS;

thrift is a network protocal, which is used to establish the connection to the server.

























 
 

Pyspark USECASE1

RDD-->Resilient Distributed Dataset(basic/root object)
Dataframes-->

RDD-->unstructured
DF-->stru/semi
unstructured-->RDD operations-->is organised as structured--->Dataframe out of the
RDD

input file:
Spark is processing framework
spark is faster than map reduce
spark and mr are processing framework

df1=spark.read.format('csv').option('delimeter',' ').load('file:///home/hadoop/sample3")

df1.show()

+--------------------+
|               value|
+--------------------+
|Spark is processi...|
|spark is faster t...|
|spark and mr are ...|
+--------------------+

df1=spark.read.format('csv').option('delimeter','##').load('file:///home/hadoop/sampe")

df1.show()-->error is thrown

rd1=spark.sparkContext.textFile("file://home/karna/Desktop/sample")

print(rd1.collect())

[u'Spark#is#processing#framework', u'spark#is#faster#than#map#reduce', u'spark#and#mr#are#processing#framework']

rd2=rd1.map(lamda x:str(x))
print(rd2.map(lambda x:type(x)))

 print(rd1.map(lambda x:str(x)))
PythonRDD[216] at RDD at PythonRDD.scala:53


print(rd2.collect())

['Spark#is#processing#framework', 'spark#is#faster#than#map#reduce', 'spark#and#mr#are#processing#framework']
>>>

rd2.count()
3

rd3=rd2.map(lambda x:x.split('##'))

print(rd3.collect())----> it is returned to the list

To find the length of each string
[['Spark', 'is', 'processing', 'framework'], ['spark', 'is', 'faster', 'than', 'map', 'reduce'], ['spark', 'and', 'mr', 'are', 'processing', 'framework']]

>>> rd3.map(lambda x:len(x)).collect()
[4, 6, 6]

>>> rd4=rd3.map(lambda x:(x[0],x[1],x[2],x[3]))

>>> rd4.collect()
[('Spark', 'is', 'processing', 'framework'), ('spark', 'is', 'faster', 'than'), ('spark', 'and', 'mr', 'are')]
>>>

>> print(rd4.map(lambda x:type(x)).collect())
[<type 'tuple'>, <type 'tuple'>, <type 'tuple'>]


df=rd4.toDF()

df1.show(()

df1=rd4.toDF(["id1","id2","id3","id4","id5"])

df1.show()

+-----+---+----------+---------+
|   _1| _2|        _3|       _4|
+-----+---+----------+---------+
|Spark| is|processing|framework|
|spark| is|    faster|     than|
|spark|and|        mr|      are|
+-----+---+----------+---------+

df.printSchema()
+-----+---+----------+---------+
|   _1| _2|        _3|       _4|
+-----+---+----------+---------+
|Spark| is|processing|framework|
|spark| is|    faster|     than|
|spark|and|        mr|      are|
+-----+---+----------+---------+


df_rsv=spark.read.format('csv').load('file://home")

df_tsv=spark.read.format('csv').option('inferSchema',True).option('delimeter','\t').
load("\home\\")

df_tsv.show(5)

df_tsv.printSchema

from pyspark.sql.types import *

sch=StructType([StructField('id',StringType()),StructField('id2',StringType()).StructField('name',StringType())]).StructField('country',StringType).StructField('year',StringType()).StructField('game',StringType()])



df_tsv=spark.read.format('csv').option('inferSchema',True).option('delimeter','\t').schema(sch).load("\home\\")

create RDD out of Dataframe
df_rdd=df_tsv.rdd===> changing to the ROW object
df_rdd

>>> rd2.collect()
[Row(_1=u'Spark', _2=u'is', _3=u'processing', _4=u'framework'), Row(_1=u'spark', _2=u'is', _3=u'faster', _4=u'than'), Row(_1=u'spark', _2=u'and', _3=u'mr', _4=u'are')]
>>> rd2.take(3)
[Row(_1=u'Spark', _2=u'is', _3=u'processing', _4=u'framework'), Row(_1=u'spark', _2=u'is', _3=u'faster', _4=u'than'), Row(_1=u'spark', _2=u'and', _3=u'mr', _4=u'are')]


df_rdd.map(lambda x:type(x)).take(10)
[<class 'pyspark.sql.types.Row'>, <class 'pyspark.sql.types.Row'>, <class 'pyspark.sql.types.Row'>]



rd4.toDF()

RDD(tuples) rd1.toDF(["c1","c2"})

RDD(ROW)-->df_rowrdd=spark.createDataFrame(df_rdd,sch)

df_rowrdd.show(5)

>>> df5=rd5.toDF()
>>> df5.show()
+-----+---+----------+---------+
|   _1| _2|        _3|       _4|
+-----+---+----------+---------+
|Spark| is|processing|framework|
|spark| is|    faster|     than|
|spark|and|        mr|      are|
+-----+---+----------+---------+

Spark,is,processing,framework
spark is faster than map reduce
spark and mr are processing framework and also computing frameworks

r1=spark.sparkContext.textFile("file:////home/karna/sample")

r2=r1.map(lamda x:(str)x)

r3=r2.map(lamda y:y.split(','))

r3.collect()

r3.map(lamda x:len(x)).collect()

r4=r3.map(lamda x:(x[0],x[1],x[2],x[3],x[4],if x[5]='' return 'nodata',x[6],x[8],x[9]))
r4.collect()

work in notepad
def handlenull(x):
  if x=='':
    return "nodata"
  else:
    return x

handlenull('python')
print(k)

python py1.py1
def handlenull(x):
result=[]
for i in range(10):
  if x[i]=='':
    return result.append("nodata")
  else:
    return result.append(x[i])

handlenull([1,2,3])
print(r)

DAG-Direct Asyclic Graph(flow of execution of RDD)