All about Kryo serialiser in Spark

-

Q. Why Kryo is fatser?

Ans: Kryo is significantly faster and more compact than Java serialization

Q. If Kryo serialiser is faster then why it is not default serialiser?

does not support all Serializable types and requires you to register the classes you’ll use in the program in advance for best performance.

So it is not used by default because:

  1. Not every java.io.Serializable is supported out of the box – if you have custom class that extends Serializable it still cannot be serialized with Kryo, unless registered.
  2. One needs to register custom classes.

The only reason Kryo is not set to default is because it requires custom registration. Although, Kryo is supported for RDD caching and shuffling, it’s not natively supported to serialize to the disk. Both the methods, saveAsObjectFile on RDD and objectFile method on SparkContext supports only java serialization.

Still?

If you need a performance boost and also need to reduce memory usage, Kryo is definitely for you. The join operations and the grouping operations are where serialization has an impact on and they usually have data shuffling. Now lesser the amount of data to be shuffled, the faster will be the operation.
Caching also have an impact when caching to disk or when data is spilled over from memory to disk.

Q. How to register class with kryo?

Kryo register a class, we simply have to pass the name of the class in the registerKryoClasses method. i.e :

.registerKryoClasses(       Array(classOf[Person], classOf[Furniture])     )

Q. Kryo Example:

Example

Lets look with a simple example to see the difference with the default Java Serialization in practical.
Starting off by registering the required classes.

//class which needs to be registeredcase class Person(name: String, age: Int)val conf = new SparkConf()    .setAppName("kyroExample")    .setMaster("local[*]")    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")    .set("spark.kryo.registrationRequired", "true")    .registerKryoClasses(      Array(classOf[Person],classOf[Array[Person]],Class.forName("org.apache.spark.internal.io.FileCommitProtocol$TaskCommitMessage"))    )val sparkContext = new SparkContext(conf)

Now, lets create an array of Person and parallelize it to make an RDD out of it and persist it in memory.

val personList: Array[Person] = (1 to 9999999)                     .map(value => Person("p"+value, value)).toArray//creating RDD of Personval rddPerson: RDD[Person] = sparkContext.parallelize(personList,5)val evenAgePerson: RDD[Person] = rddPerson.filter(_.age % 2 == 0)//persisting evenAgePerson RDD into memoryevenAgePerson.persist(StorageLevel.MEMORY_ONLY_SER)evenAgePerson.take(50).foreach(x=>println(x.name,x.age))

 

Popular posts from this blog

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

Complex SQL: fetch the users who logged in consecutively 3 or more times (lead perfect example)

-
From the login_details table, fetch the users who logged in consecutively 3 or more times. Table Name : LOGIN_DETAILS Approach : We need to fetch users who have appeared 3 or more times consecutively in login details table. There is a window function which can be used to fetch data from the following record. Use that window function to compare the user name in current row with user name in the next row and in the row following the next row. If it matches then fetch those records. –Table Structure: drop table login_details; create table login_details( login_id int primary key, user_name varchar(50) not null, login_date date); delete from login_details; insert into login_details values (101, ‘Michael’, current_date), (102, ‘James’, current_date), (103, ‘Stewart’, current_date+1), (104, ‘Stewart’, current_date+1), (105, ‘Stewart’, current_date+1), (106, ‘Michael’, current_date+2), (107, ‘Michael’, cur...
Read more

Credit Card Data Analysis using PySpark (how to use auto broadcast join after disabling it)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Application: Here we will disable auto broadcast and then join 2 data frames and will see execution plan which should use broadcast join Solution: from pyspark.sql import SparkSession from pyspark.sql.functions import broadcast spark = SparkSession.builder.master(‘local[2]’)\ .appName(‘RDD_Methods_Examples’)\ .getOrCreate() print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”)) spark.conf.set(“spark.sql.autoBroadcastJoinThreshold”, -1) pr...
Read more