JsonStreamConsumerExample.scala

package imaifactory.examples

import org.apache.spark.streaming._
import org.apache.spark.streaming.kinesis._
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.storage.StorageLevel

object JsonStreamConsumerExample extends App {


  val appName = "APP_NAME"
  val streamName = "STREAM_NAME"
  val endpointUrl = "https://kinesis.REGION.amazonaws.com"
  val region = "REGION"
  val position = InitialPositionInStream.LATEST

  val conf = new SparkConf().setAppName(appName).setMaster("local[*]")
  val ssc = new StreamingContext(conf, Seconds(10))
  val spark = SparkSession.builder.config(conf).getOrCreate()

  val kinesisStream = KinesisUtils.createStream(
    ssc, appName, streamName, endpointUrl,
    region, position, Seconds(10), StorageLevel.MEMORY_AND_DISK_2)

  val dataStream = kinesisStream.foreachRDD(rdd => {

    val df = spark.read.json(rdd.map(record => { new String(record)}))

    if(df.count() > 0) {
      df.createOrReplaceTempView("bx1")
      spark.sql("select * from bx1").show
    }

  })

  ssc.start()           
  ssc.awaitTermination()  
}