Powered By Blogger

Sunday, October 6, 2019

Using Date Format in dataframe


scala> import org.apache.spark.sql.functions.to_date

import org.apache.spark.sql.functions.to_date

scala> val dateFormat = "yyyy-dd-MM"
dateFormat: String = yyyy-dd-MM

scala> val cleanDateDF = spark.range(1).select(
     | to_date(lit("2017-12-11"), dateFormat).alias("date"),
     | to_date(lit("2017-20-12"), dateFormat).alias("date2"))
cleanDateDF: org.apache.spark.sql.DataFrame = [date: date, date2: date]

scala> cleanDateDF.show(2)
+----------+----------+
|      date|     date2|
+----------+----------+
|2017-11-12|2017-12-20|
+----------+----------+





Spark the Definitive Guide is having good details about manipulating Hadoop

Converting String field to Date in Dataframe


val sourceDF = spark.createDF(
  List(
    (1, "2013-01-30"),
    (2, "2012-01-01")
  ), List(
    ("person_id", IntegerType, true),
    ("birth_date", StringType, true)
  )
).withColumn(
  "birth_date",
  col("birth_date").cast("date")
)

sourceDF.show()
sourceDF.printSchema()


https://mungingdata.com/apache-spark/dates-times/

Saturday, October 5, 2019

Converting Dataframe String type to Integer

acbc32a57245:bin z002qhl$ ./spark-shell
19/10/06 12:05:28 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://192.168.1.2:4040
Spark context available as 'sc' (master = local[*], app id = local-1570343764090).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.1
      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_191)
Type in expressions to have them evaluated.
Type :help for more information.

scala> import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.IntegerType

scala> import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StringType

scala> import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructType

scala> import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StructField

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

scala> import scala.collection.JavaConversions._
import scala.collection.JavaConversions._

scala>     val data = Seq(("Java", "20000"), ("Python", "100000"), ("Scala", "3000"))
data: Seq[(String, String)] = List((Java,20000), (Python,100000), (Scala,3000))

scala>     val tableColumns = List(("order_number", "string"), ("external_order_number", "string"))
tableColumns: List[(String, String)] = List((order_number,string), (external_order_number,string))

scala>     var schema = new StructType
schema: org.apache.spark.sql.types.StructType = StructType()

scala>     for (i <- p="" tablecolumns="">     |       schema = schema.add(i._1, i._2)

scala>       val rowData = data.map(attributes => Row(attributes._1, attributes._2))
rowData: Seq[org.apache.spark.sql.Row] = List([Java,20000], [Python,100000], [Scala,3000])

scala>       var dfFromData4 = spark.createDataFrame(rowData,schema)
dfFromData4: org.apache.spark.sql.DataFrame = [order_number: string, external_order_number: string]

scala>       dfFromData4.show(2)
+------------+---------------------+
|order_number|external_order_number|
+------------+---------------------+
|        Java|                20000|
|      Python|               100000|
+------------+---------------------+
only showing top 2 rows


scala>

scala> val dfFromData5 = dfFromData4.withColumn("external_order_number", 'external_order_number cast "int")
dfFromData5: org.apache.spark.sql.DataFrame = [order_number: string, external_order_number: int]

scala> dfFromData5.printSchema
root
 |-- order_number: string (nullable = true)
 |-- external_order_number: integer (nullable = true)
<- p="" tablecolumns="">
<- p="" tablecolumns="">
<- p="" tablecolumns="">
<- p="" tablecolumns="">
<- p="" tablecolumns="">---------
<- p="" tablecolumns="">
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.Row
import scala.collection.JavaConversions._
    val data = Seq(("Java", "20000"), ("Python", "100000"), ("Scala", "3000"))
    val tableColumns = List(("order_number", "string"), ("external_order_number", "string"))
    var schema = new StructType
    for (i <- p="" tablecolumns="">      schema = schema.add(i._1, i._2)
      val rowData = data.map(attributes => Row(attributes._1, attributes._2))
      var dfFromData4 = spark.createDataFrame(rowData,schema)
      dfFromData4.show(2)


val dfFromData6 =dfFromData4.withColumn("external_order_number", dfFromData4("external_order_number").cast("int"))
<- p="" tablecolumns="">
dfFromData6.printSchema

Message to RDD to Dataframe example

import spark.implicits._
val columns = Seq("language","users_count")
val data = Seq(("Java", "20000"), ("Python", "100000"), ("Scala", "3000"))
val rdd = spark.sparkContext.parallelize(data)
val dfFromRDD1 = rdd.toDF()
dfFromRDD1.printSchema()
val dfFromRDD1 = rdd.toDF("language","users_count")
dfFromRDD1.printSchema()

val dfFromRDD2 = spark.createDataFrame(rdd).toDF(columns:_*)

import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.Row
val schema = StructType(columns
  .map(fieldName => StructField(fieldName, StringType)))
val rowRDD = rdd.map(attributes => Row(attributes._1, attributes._2))
val dfFromRDD3 = spark.createDataFrame(rowRDD,schema)

import spark.implicits._
val dfFromData1 = data.toDF()


import scala.collection.JavaConversions._

val rowData = data.map(attributes => Row(attributes._1, attributes._2))
var dfFromData4 = spark.createDataFrame(rowData,schema)

Message to Dataframe using Schema

scala> acbc32a57245:bin z002qhl$ ./spark-shell
19/10/06 11:40:52 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://192.168.1.2:4040
Spark context available as 'sc' (master = local[*], app id = local-1570342289541).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.1
      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_191)
Type in expressions to have them evaluated.
Type :help for more information.

scala> import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.types.IntegerType

scala> import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StringType

scala> import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructType

scala> import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StructField

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

scala> import scala.collection.JavaConversions._
import scala.collection.JavaConversions._

scala>     val data = Seq(("Java", 20000), ("Python", 100000), ("Scala", 3000))
data: Seq[(String, Int)] = List((Java,20000), (Python,100000), (Scala,3000))

scala>     val tableColumns = List(("order_number", "string"), ("external_order_number", "integer"))
tableColumns: List[(String, String)] = List((order_number,string), (external_order_number,integer))

scala>     var schema = new StructType
schema: org.apache.spark.sql.types.StructType = StructType()

scala>     for (i <- p="" tablecolumns="">     |       schema = schema.add(i._1, i._2)

scala>       val rowData = data.map(attributes => Row(attributes._1, attributes._2))
rowData: Seq[org.apache.spark.sql.Row] = List([Java,20000], [Python,100000], [Scala,3000])

scala>       var dfFromData4 = spark.createDataFrame(rowData,schema)
dfFromData4: org.apache.spark.sql.DataFrame = [order_number: string, external_order_number: int]

scala>       dfFromData4.show(2)
+------------+---------------------+
|order_number|external_order_number|
+------------+---------------------+
|        Java|                20000|
|      Python|               100000|
+------------+---------------------+
only showing top 2 rows

creating dataframe in spark shell

acbc32a57245:bin z002qhl$ pwd
/Users/basan/Documents/Spark/spark-2.4.1-bin-hadoop2.7/bin
acbc32a57245:bin z002qhl$ ./spark-s
spark-shell   spark-sql     spark-submit

acbc32a57245:bin basan$ ./spark-shell
19/10/06 08:43:21 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://192.168.1.4:4040
Spark context available as 'sc' (master = local[*], app id = local-1570331637922).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.1
      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_191)
Type in expressions to have them evaluated.
Type :help for more information.

scala> val data = List(("James ","","Smith","36636","M",60000),
     |         ("Michael ","Rose","","40288","M",70000),
     |         ("Robert ","","Williams","42114","",400000),
     |         ("Maria ","Anne","Jones","39192","F",500000),
     |         ("Jen","Mary","Brown","","F",0))
data: List[(String, String, String, String, String, Int)] = List(("James ","",Smith,36636,M,60000), ("Michael ",Rose,"",40288,M,70000), ("Robert ","",Williams,42114,"",400000), ("Maria ",Anne,Jones,39192,F,500000), (Jen,Mary,Brown,"",F,0))

scala> val cols = Seq("first_name","middle_name","last_name","dob","gender","salary")
cols: Seq[String] = List(first_name, middle_name, last_name, dob, gender, salary)

scala> val df = spark.createDataFrame(data).toDF(cols:_*)
df: org.apache.spark.sql.DataFrame = [first_name: string, middle_name: string ... 4 more fields]

scala> df.show(2)
+----------+-----------+---------+-----+------+------+
|first_name|middle_name|last_name|  dob|gender|salary|
+----------+-----------+---------+-----+------+------+
|    James |           |    Smith|36636|     M| 60000|
|  Michael |       Rose|         |40288|     M| 70000|
+----------+-----------+---------+-----+------+------+
only showing top 2 rows


scala> val df4 = df.select(col("*"),
     |       expr("case when gender = 'M' then 'Male' " +
     |                        "when gender = 'F' then 'Female' " +
     |                        "else 'Unknown' end").alias("new_gender"))
df4: org.apache.spark.sql.DataFrame = [first_name: string, middle_name: string ... 5 more fields]

scala> df4.show(2)
+----------+-----------+---------+-----+------+------+----------+
|first_name|middle_name|last_name|  dob|gender|salary|new_gender|
+----------+-----------+---------+-----+------+------+----------+
|    James |           |    Smith|36636|     M| 60000|      Male|
|  Michael |       Rose|         |40288|     M| 70000|      Male|
+----------+-----------+---------+-----+------+------+----------+
only showing top 2 rows


scala> val df3 = df.withColumn("new_gender2",
     |       expr("case when gender = 'M' then 'Male' " +
     |                        "when gender = 'F' then 'Female' " +
     |                        "else 'Unknown' end"))
df3: org.apache.spark.sql.DataFrame = [first_name: string, middle_name: string ... 5 more fields]

scala> df3.show(2)
+----------+-----------+---------+-----+------+------+-----------+
|first_name|middle_name|last_name|  dob|gender|salary|new_gender2|
+----------+-----------+---------+-----+------+------+-----------+
|    James |           |    Smith|36636|     M| 60000|       Male|
|  Michael |       Rose|         |40288|     M| 70000|       Male|
+----------+-----------+---------+-----+------+------+-----------+
only showing top 2 rows


scala>

scala> val df3 = df.withColumn("geneder",
     |       expr("case when gender = 'M' then 'Male' " +
     |                        "when gender = 'F' then 'Female' " +
     |                        "else 'Unknown' end"))
df3: org.apache.spark.sql.DataFrame = [first_name: string, middle_name: string ... 5 more fields]

scala> df3.show(2)
+----------+-----------+---------+-----+------+------+-------+
|first_name|middle_name|last_name|  dob|gender|salary|geneder|
+----------+-----------+---------+-----+------+------+-------+
|    James |           |    Smith|36636|     M| 60000|   Male|
|  Michael |       Rose|         |40288|     M| 70000|   Male|
+----------+-----------+---------+-----+------+------+-------+
only showing top 2 rows


scala>

scala> val df3 = df.withColumn("gender",
     |       expr("case when gender = 'M' then 'Male' " +
     |                        "when gender = 'F' then 'Female' " +
     |                        "else 'Unknown' end"))
df3: org.apache.spark.sql.DataFrame = [first_name: string, middle_name: string ... 4 more fields]

scala> df3.show(2)
+----------+-----------+---------+-----+------+------+
|first_name|middle_name|last_name|  dob|gender|salary|
+----------+-----------+---------+-----+------+------+
|    James |           |    Smith|36636|  Male| 60000|
|  Michael |       Rose|         |40288|  Male| 70000|
+----------+-----------+---------+-----+------+------+
only showing top 2 rows


Wednesday, October 2, 2019

Add jar to spark classpath

scala> :require /Users/basan/.ivy2/cache/org.apache.kafka/kafka-clients/jars/kafka-clients-0.10.2.2.jar
Added '/Users/basan/.ivy2/cache/org.apache.kafka/kafka-clients/jars/kafka-clients-0.10.2.2.jar' to classpath.


scala> import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.consumer.ConsumerRecord

scala>      val record1 : ConsumerRecord[String, String] = new ConsumerRecord("topic1",0,100,"Key1", "value1")
record1: org.apache.kafka.clients.consumer.ConsumerRecord[String,String] = ConsumerRecord(topic = topic1, partition = 0, offset = 100, NoTimestampType = -1, checksum = -1, serialized key size = -1, serialized value size = -1, key = Key1, value = value1)

scala>     val record2 : ConsumerRecord[String, String] = new ConsumerRecord("topic1",0,100,"Key2", "value2")
record2: org.apache.kafka.clients.consumer.ConsumerRecord[String,String] = ConsumerRecord(topic = topic1, partition = 0, offset = 100, NoTimestampType = -1, checksum = -1, serialized key size = -1, serialized value size = -1, key = Key2, value = value2)

scala>     record1.value()
res0: String = value1