ashrithr

Introduction to Kafka

Kafka acts as a kind of write-ahead log (WAL) that records messages to a persistent store (disk) and allows subscribers to read and apply these changes to their own stores in a system appropriate time-frame.

Terminology:

• Producers send messages to brokers
• Consumers read messages from brokers
• Messages are sent to a topic

satendrakumar

object play {
  println("Welcome to the Scala worksheet")
  
  case class User(id:Int, name:String)
  def getUser:User = null
  def processUser(u:User):String= u.name
 Option(getUser)
  
  Option(getUser).fold("Sky")(u=>processUser(u))
  

satendrakumar

object GenericTypeConverter extends App{

  def convert[T: ClassTag](value:Any): Option[T] = {
    val ct = implicitly[ClassTag[T]]
        value match {
          case ct(x) => Some(x)
          case _     => None
        }
    }

satendrakumar

Prelude> let forall :: (a -> Bool) -> [a] -> Bool ; forall f []= True; forall f (x:xs) = f x  &&  forall f  xs 
Prelude> forall even  [2,4,6] 
True
Prelude> forall even [2,4,5]
False
Prelude> let forall :: (a -> Bool) -> [a] -> Bool ; forall f []= True; forall f (x:xs) = if (f x)  then forall f  xs else False
Prelude> forall even [2,4,5]
False
Prelude> forall even  [2,4,6] 
True

satendrakumar

import org.json4s._
import org.json4s.native.{JsonMethods, Serialization}

object JsonUtility {

  implicit val formats = DefaultFormats

  def write[T <: AnyRef](value: T): String = Serialization.write(value)

satendrakumar

import org.slf4j.{Logger, LoggerFactory}

trait Logging {

  protected val logger: Logger = LoggerFactory.getLogger(this.getClass())

  protected def debug(message: String): Unit = logger.debug(message)

  protected def debug(message: String, exception: Throwable): Unit = logger.debug(message, exception)

satendrakumar

import com.univocity.parsers.csv.{CsvParser, CsvParserSettings}


class CSVParser(delimiter: Char = ',') {

  private val parser = {
    val settings = new CsvParserSettings
    val format = settings.getFormat
    format.setLineSeparator("\n")
    format.setDelimiter(delimiter)

satendrakumar

import java.io.{BufferedReader, File, InputStreamReader}
import javax.annotation.concurrent.NotThreadSafe

import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.mapred.{FileSplit, TextInputFormat}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.HadoopRDD
import org.apache.spark.{SparkConf, SparkContext}

@NotThreadSafe

satendrakumar

object FoldLeft extends App {

  def reverse(list: List[Int]): List[Int] =
    list.foldLeft[List[Int]](Nil)((acc, element) => element :: acc)

  def dedupe(list: List[Int]): List[Int] = {
    list.foldLeft[List[Int]](Nil)((acc, element) => if (acc.contains(element)) acc else acc :+ element)
  }

satendrakumar

import java.io._

object Loop1 extends App {
 //Java way 
  val inputStream: BufferedInputStream = new BufferedInputStream(new FileInputStream("input.csv"))
  val outputStream = new BufferedOutputStream(new FileOutputStream("output.csv"))
  val buffer = new Array[Byte](32 * 1024)
  var bytesRead: Int = inputStream.read(buffer)
  while (bytesRead > 0) {
    println("writing.......")