spark salting technique explained

 the fallowing is for spark salting technique

var df1 = Seq((1,"a"),(2,"b"),(1,"c"),(1,"x"),(1,"y"),(1,"g"),(1,"k"),(1,"u"),(1,"n")).toDF("ID","NAME")

df1.createOrReplaceTempView("fact")

var df2 = Seq((1,10),(2,30),(3,40)).toDF("ID","SALARY")

df2.createOrReplaceTempView("dim")

val salted_df1 = spark.sql("""select concat(ID, '_', FLOOR(RAND(123456)*19)) as salted_key, NAME from fact """)

salted_df1.createOrReplaceTempView("salted_fact")

# remember that explode array should start with 0 and end with whatever the values we mentioned in the above flor and rand

val exploded_dim_df = spark.sql(""" select ID, SALARY, explode(array(0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19)) as salted_key from dim""")

//val exploded_dim_df = spark.sql(""" select ID, SALARY, explode(array(0 to 19)) as salted_key from dim""")

exploded_dim_df.createOrReplaceTempView("salted_dim")

val result_df = spark.sql("""select split(fact.salted_key, '_')[0] as ID, dim.SALARY 

            from salted_fact fact 

            LEFT JOIN salted_dim dim 

            ON fact.salted_key = concat(dim.ID, '_', dim.salted_key) """)

display(result_df)

How to set timezone to UTC in Apache Spark?

 Now you can use:

spark.conf.set("spark.sql.session.timeZone", "UTC")

Since https://issues.apache.org/jira/browse/SPARK-18936 in 2.2.0

EDIT:

Additionally I set my default TimeZone to UTC to avoid implicit conversions

TimeZone.setDefault(TimeZone.getTimeZone("UTC"))

Otherwise you will get implicit conversions from your default Timezone to UTC when no Timezone information is present in the Timestamp you're converting

Example:

val rawJson = """ {"some_date_field": "2018-09-14 16:05:37"} """

val dsRaw = sparkJob.spark.createDataset(Seq(rawJson))

val output =
  dsRaw
    .select(
      from_json(
        col("value"),
        new StructType(
          Array(
            StructField("some_date_field", DataTypes.TimestampType)
          )
        )
      ).as("parsed")
    ).select("parsed.*")

If my default TimeZone is Europe/Dublin which is GMT+1 and Spark sql session timezone is set to UTC, Spark will assume that "2018-09-14 16:05:37" is in Europe/Dublin TimeZone and do a conversion (result will be "2018-09-14 15:05:37")

PySpark

spark = pyspark.sql.SparkSession \
    .Builder()\
    .appName('test') \
    .master('local') \
    .config('spark.driver.extraJavaOptions', '-Duser.timezone=GMT') \
    .config('spark.executor.extraJavaOptions', '-Duser.timezone=GMT') \
    .config('spark.sql.session.timeZone', 'UTC') \
    .getOrCreate()




Spark Performance Tuning Optimisation

 Apache Spark Performance Tuning and Optimizations for Big Datasets

Spark performance tuning from the trenches

Spark Optimisation Techniques

Apache Spark Optimization Techniques and Tuning

Fine Tuning and Enhancing Performance of Apache Spark Jobs

Optimizing the Skew in Spark

Spark Job Optimization: Dealing with Data Skew

Spark Scala Notes

Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.SparkSession

 problem: 

Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.SparkSession
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:349)
at java.lang.ClassLoader.loadClass

Solution:

change the scope of all the spark dependencies from provided to compile

Exception in thread “main” java.lang.UnsatisfiedLinkError: org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Z

 In addition to other solutions, Please download winutil.exe and hadoop.dll and add to $HADOOP_HOME/bin. It works for me.

https://github.com/steveloughran/winutils/tree/master/hadoop-2.7.1/bin

Note: I'm using hadoop-2.7.3 version 

 

How do implicit work in Spark/Scala ?

Q: 

Here is a sample Spark code, which converts the Seq to the Dataset:

import spark.implicits._
val s = Seq(1, 2, 3, 4)
val ds = s.toDS()

The Scala Seq does not have the toDS method, it comes from Spark implicits. How is the Dataset created here?

A:

Scala has a way to add methods to existing classes, like extension method in Kotlin (and C# as I remember), but does it in a different way, through implicits.

To add the method to existing class, you first create implicit class:

object StringImplicits {
  implicit class StringUtils(s: String) {
    def someCoolMethod = println("Yooo")
  }
}

object Application extends App {
    import StringImplicits._
    val s = "Hello"
    s.someCoolMethod
}

You import this StringUtils and can call someCoolMethod on instance of String

Notice that StringUtils class takes String as a constuctor param.

When calling some method on String, scala compiler first looks this method in String class.

If it does not find it, it will look imported implicit classes which take String param.

If found, it calls the method from that class.

If no such class found, it will raise the error.

Create tuple in spark

scala> val rdd = sc.parallelize(for {

     |     x <- 1 to 3

     |     y <- 1 to 2

     | } yield (x, y), 8)

rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[13] at parallelize at <console>:24

scala> rdd.collect

res8: Array[(Int, Int)] = Array((1,1), (1,2), (2,1), (2,2), (3,1), (3,2))

The authorization mechanism you have provided is not supported. Please use AWS4-HMAC-SHA256 - Databricks

Solution:

Step1: 

Creating user bucket with region (US West (Oregon))

Now Testing the Application.

reliable receiver VS unreliable receiver

Receiver's API

Among technical considerations, receivers can be built by extending abstract class Receiver. Two methods must be implemented: onStart() and onStop(), respectively used to, start and stop data consumption. The first method can contain all necessary initialization for data reading, such as opening connections, creating new threads etc.

But be careful, onStart() must be non-blocking. Thus to be able to properly start and stop receiver, data retrieval should be implement in new thread. Otherwise, even if streaming context exceeds its timeout, it won't be stopped because of blocking onStart() method.

Data is retrieved in new thread created and started by onStart() method but how it's sent to Spark ? Actually there are 2 methods to move data from receiver to Spark context, both used in two different types of receivers. Before talking about these methods, let's explain what are receiver types:

• reliable receiver - this receiver acknowledges data sources when data is received and replicated successfully in Spark storage.
• unreliable receiver - in this situation, the ack is not sent to the source.

Under-the-hood, the difference between them comes from the method used to send data to Spark. For the case of reliable receiver, data is sent through store(...) method taking in parameter collection-like objects (Iterator, ByteBuffer or ArrayBuffer). It's a blocking method which doesn't return as long as Spark doesn't notify receiver about successful data save. After returning, receiver can acknowledge source about data reception.

In the other side, unreliable receiver uses store(...) method taking in parameter a single object. This method is not blocking but it doesn't sent data immediately to Spark. Instead of that, it keeps data in memory and sends it as batch to Spark after accumulating some number of items.

Sometimes receiver can fail. In this case, receiver can be restarted through restart(...) method. However, the restart is not immediate, it's only scheduled. The restart execution consists on calling onStop() and onStart() methods subsequently withint configured delay (spark.streaming.receiverRestartDelay property).

Set log level in spark-shell in spark REPL Environment

Follow the below steps .

1. Add the Logger 

import org.apache.log4j.Logger

2. Add the Level

import org.apache.log4j.Level
3.Set the Log Level (Here we are setting the Level to "org" package)
Logger.getLogger("org").setLevel(Level.INFO)  (We can add any Level Like  DEBUG,ERROR,TRACE,OFF )

How to use braodcast variable in spark

Broadcast Variables

Broadcast variables allow the programmer to keep a read-only variable cached on each machine rather than shipping a copy of it with tasks. They can be used, for example, to give every node a copy of a large input dataset in an efficient manner. Spark also attempts to distribute broadcast variables using efficient broadcast algorithms to reduce communication cost.

Spark actions are executed through a set of stages, separated by distributed “shuffle” operations. Spark automatically broadcasts the common data needed by tasks within each stage. The data broadcasted this way is cached in serialized form and deserialized before running each task. This means that explicitly creating broadcast variables is only useful when tasks across multiple stages need the same data or when caching the data in deserialized form is important.

Broadcast variables are created from a variable v by calling SparkContext.broadcast(v). The broadcast variable is a wrapper around v, and its value can be accessed by calling the value method. The code below shows this:

import java.nio.charset.CodingErrorAction

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext

import scala.io.{Codec, Source}

object AverageMovieRatings {

    def mapToTuple(line: String): (Int, (Float, Int)) = {
        val fields = line.split(',')
        return (fields(1).toInt, (fields(2).toFloat, 1))
    }

    def loadMovieNames(): Map[Int, String] = {

        // Handle character encoding issues
        implicit val codec = Codec("UTF-8")
        codec.onMalformedInput(CodingErrorAction.REPLACE)
        codec.onUnmappableCharacter(CodingErrorAction.REPLACE)

        var movieNames: Map[Int, String] = Map()

        // Read lines from movies.csv into Iterator. Drop the first (header) row.
        val lines = Source.fromFile("/tmp/ml-latest-small/movies.csv").getLines().drop(1)
        for (line <- lines) {
            val fields = line.split(',')
            movieNames += (fields(0).toInt -> fields(1))
        }
        return movieNames
    }

    def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.ERROR)

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

        // Broadcast the movie names
        val names = loadMovieNames()

        // Read a text file
        var data = sc.textFile("/tmp/ml-latest-small/ratings.csv")

        // Extract the first row which is the header
        val header = data.first();

        // Filter out the header from the dataset
        data = data.filter(row => row != header)

        val result = data.map(mapToTuple)
          .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2))
          .map(x => (x._1, x._2._1 / x._2._2))
          .sortBy(_._2, false)
          .map(x => (names(x._1), x._2))
          .collect()

        result.foreach(println)
    }
}