title: How to create a KinesisSink for Spark Structural Streaming
body: <p>I am using Spark 2.2 on Databricks and trying to implement a Kinesis sink to write from Spark to a Kinesis stream.</p>
  
  <p>I am using the following provide sample from here <a href="https://docs.databricks.com/_static/notebooks/structured-streaming-kinesis-sink.html" rel="nofollow noreferrer">https://docs.databricks.com/_static/notebooks/structured-streaming-kinesis-sink.html</a></p>
  
  <pre><code>/**
  * A simple Sink that writes to the given Amazon Kinesis `stream` in the given    `region`. For authentication, users may provide
  * `awsAccessKey` and `awsSecretKey`, or use IAM Roles when launching their cluster.
  *
  * This Sink takes a two column Dataset, with the columns being the `partitionKey`, and the `data` respectively.
  * We will buffer data up to `maxBufferSize` before flushing to Kinesis in order to reduce cost.
  */
  class KinesisSink(
      stream: String,
      region: String,
      awsAccessKey: Option[String] = None,
      awsSecretKey: Option[String] = None) extends ForeachWriter[(String, Array[Byte])] { 
  
   // Configurations
   private val maxBufferSize = 500 * 1024 // 500 KB
  
   private var client: AmazonKinesis = _
   private val buffer = new ArrayBuffer[PutRecordsRequestEntry]()
   private var bufferSize: Long = 0L
  
   override def open(partitionId: Long, version: Long): Boolean = {
     client = createClient
     true
   }
  
   override def process(value: (String, Array[Byte])): Unit = {
     val (partitionKey, data) = value
     // Maximum of 500 records can be sent with a single `putRecords` request
     if ((data.length + bufferSize &gt; maxBufferSize &amp;&amp; buffer.nonEmpty) ||    buffer.length == 500) {
    flush()
  }
  buffer += new PutRecordsRequestEntry().withPartitionKey(partitionKey).withData(ByteBuffer.wrap(data))
  bufferSize += data.length
   }
  
   override def close(errorOrNull: Throwable): Unit = {
     if (buffer.nonEmpty) {
       flush()
     }
     client.shutdown()
   }
  
   /** Flush the buffer to Kinesis */
   private def flush(): Unit = {
     val recordRequest = new PutRecordsRequest()
       .withStreamName(stream)
       .withRecords(buffer: _*)
  
     client.putRecords(recordRequest)
     buffer.clear()
     bufferSize = 0
   }
  
   /** Create a Kinesis client. */
   private def createClient: AmazonKinesis = {
     val cli = if (awsAccessKey.isEmpty || awsSecretKey.isEmpty) {
       AmazonKinesisClientBuilder.standard()
         .withRegion(region)
         .build()
     } else {
       AmazonKinesisClientBuilder.standard()
         .withRegion(region)
         .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(awsAccessKey.get, awsSecretKey.get)))
         .build()
     }
     cli
   }
  }
  </code></pre>
  
  <p>I then implement the KinesisSink class using</p>
  
  <pre><code>val kinesisSink = new KinesisSink("us-east-1", "MyStream", Option("xxx..."), Option("xxx..."))
  </code></pre>
  
  <p>Finally i create a stream using this sink. This KinesisSink takes a two column Dataset, with the columns being the <code>partitionKey</code>, and the <code>data</code> respectively.</p>
  
  <pre><code>case class MyData(partitionKey: String, data: Array[Byte])
  
  val newsDataDF = kinesisDF
     .selectExpr("apinewsseqid", "fullcontent").as[MyData]
     .writeStream
     .outputMode("append")
     .foreach(kinesisSink)
     .start
  </code></pre>
  
  <p>but i am still getting the following error</p>
  
  <pre><code>error: type mismatch;
  found   : KinesisSink
  required: org.apache.spark.sql.ForeachWriter[MyData]
     .foreach(kinesisSink)
  </code></pre>