title: Receive a Kafka message through Spark Streaming and delete Phoenix/HBase data
body: <p>In my project, I have the current workflow:
  Kafka message =&gt; Spark Streaming/processing =&gt; Insert/Update to HBase and/or Phoenix</p>
  <p>Both the Insert and Update operation works with HBase directly or through Phoenix (I tested both case).</p>
  <p>Now I would like to delete data in HBase/Phoenix if I receive a specific message on Kafka. I did not find any clues/documentation on how I could do that while streaming.</p>
  <p>I have found a way to delete data in &quot;static&quot;/&quot;batch&quot; mode with both HBase and Phoenix, but the same code does not work when on streaming (there is no error though, the data is simply not deleted).</p>
  <p>Here is how we tried the delete part (we first create a parquet file on which we make a &quot;readStream&quot; to start a &quot;fake&quot; stream):</p>
  <p><em><strong>Main Object:</strong></em></p>
  <pre><code>import org.apache.spark.sql.{Row, SparkSession} import org.apache.spark.sql.functions.{col, concat_ws} import org.apache.spark.sql.streaming.DataStreamWriter
  
  object AppMain{   def main(args: Array[String]): Unit = {
      val spark : SparkSession = SparkSession.builder().getOrCreate()
  
  
      import spark.implicits._
      //algo_name, partner_code, site_code, indicator
      val df = Seq((&quot;FOO&quot;,&quot;FII&quot;),
        (&quot;FOO&quot;,&quot;FUU&quot;)
      ).toDF(&quot;col_1&quot;,&quot;col_2&quot;)
  
      df.write.mode(&quot;overwrite&quot;).parquet(&quot;testParquetStream&quot;)
      df.printSchema()
      df.show(false)
  
      val dfStreaming = spark.readStream.schema(df.schema).parquet(&quot;testParquetStream&quot;)
      dfStreaming.printSchema()
      val dfWithConcat = dfStreaming.withColumn(&quot;row&quot;, concat_ws(&quot;\u0000&quot; , col(&quot;col_1&quot;),col(&quot;col_2&quot;))).select(&quot;row&quot;)
  
      // using delete class
      val withDeleteClass : WithDeleteClass = new WithDeleteClass(spark, dfWithConcat)
      withDeleteClass.delete_data()
      // using JDBC/Phoenix 
      //val withJDBCSink : WithJDBCSink = new WithJDBCSink(spark, dfStreaming)
      //withJDBCSink.delete_data() }
  </code></pre>
  <p><em><strong>WithJDBCSink class</strong></em></p>
  <pre><code>import org.apache.spark.sql.{DataFrame, Row, SparkSession}
  import org.apache.spark.sql.streaming.DataStreamWriter
  
  class WithJDBCSink (spark : SparkSession,dfToDelete : DataFrame){
  
    val df_writer = dfToDelete.writeStream
  
    def delete_data():Unit = {
      val writer = new JDBCSink()
      df_writer.foreach(writer).outputMode(&quot;append&quot;).start()
      spark.streams.awaitAnyTermination()
    }
  
  }
  </code></pre>
  <p><em><strong>JDBCSink class</strong></em></p>
  <pre><code>import java.sql.{DriverManager, PreparedStatement, Statement}
  
  class JDBCSink() extends org.apache.spark.sql.ForeachWriter[org.apache.spark.sql.Row] {
  
    val quorum = &quot;phoenix_quorum&quot;
    var connection: java.sql.Connection = null
    var statement: Statement = null
    var ps : PreparedStatement= null
    def open(partitionId: Long, version: Long): Boolean = {
      connection = DriverManager.getConnection(s&quot;jdbc:phoenix:$quorum&quot;)
      statement = connection.createStatement()
      true
    }
  
    def process(row: org.apache.spark.sql.Row): Unit = {
      //-----------------------Method 1
      connection = DriverManager.getConnection(s&quot;jdbc:phoenix:$quorum&quot;)
      val query = s&quot;DELETE from TABLE_NAME WHERE key_1 = 'val1' and key_2 = 'val2'&quot;
      statement = connection.createStatement()
      statement.executeUpdate(query)
      connection.commit()
  
      //-----------------------Method 2
      //val query2 = s&quot;DELETE from TABLE_NAME WHERE key_1 = ? and key_2 = ?&quot;
      //connection = DriverManager.getConnection(s&quot;jdbc:phoenix:$quorum&quot;)
      //ps = connection.prepareStatement(query2)
      //ps.setString(1, &quot;val1&quot;)
      //ps.setString(2, &quot;val2&quot;)
      //ps.executeUpdate()
      //connection.commit()
    }
  
    def close(errorOrNull: Throwable): Unit = {
      connection.commit()
      connection.close
    }
  }
  </code></pre>
  <p><em><strong>WithDeleteClass</strong></em></p>
  <pre><code>import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
  import org.apache.spark.sql.{DataFrame, SparkSession}
  import org.apache.hadoop.hbase.client.{ConnectionFactory, Delete, HTable, RetriesExhaustedWithDetailsException, Row, Table}
  import org.apache.hadoop.hbase.ipc.CallTimeoutException
  import org.apache.hadoop.hbase.util.Bytes
  import org.apache.log4j.Logger
  import org.apache.spark.rdd.RDD
  import org.apache.spark.sql.SparkSession
  import java.util
  import org.apache.spark.sql.streaming.DataStreamWriter
  import org.apache.spark.sql.types.{StringType, StructField, StructType}
  import org.apache.spark.storage.StorageLevel
  
  import java.sql.Connection
  import java.sql.DriverManager
  import java.sql.ResultSet
  import java.sql.SQLException
  import java.sql.Statement
  
  class WithDeleteClass (spark : SparkSession, dfToDelete : DataFrame){
    val finalData = dfToDelete
    var df_writer = dfToDelete.writeStream
    var dfWriter = finalData.writeStream.outputMode(&quot;append&quot;).format(&quot;console&quot;).start()
  
    def delete_data(): Unit = {
      deleteDataObj.open()
      df_writer.foreachBatch((output : DataFrame, batchId : Long) =&gt;
        deleteDataObj.process(output)
      )
      df_writer.start()
  
    }
  
    object deleteDataObj{
      var quorum = &quot;hbase_quorum&quot;
      var zkPort = &quot;portNumber&quot;
      var hbConf = HBaseConfiguration.create()
      hbConf.set(&quot;hbase.zookeeper.quorum&quot;, quorum)
      hbConf.set(&quot;hbase.zookeeper.property.clientPort&quot;, zkPort)
      var tableName: TableName = TableName.valueOf(&quot;TABLE_NAME&quot;)
      var conn = ConnectionFactory.createConnection(hbConf)
      var table: Table = conn.getTable(tableName)
      var hTable: HTable = table.asInstanceOf[HTable]
      def open() : Unit = {
      }
      def process(df : DataFrame) : Unit = {
        val rdd : RDD[Array[Byte]] = df.rdd.map(row =&gt; Bytes.toBytes(row(0).toString))
        val deletions : util.ArrayList[Delete] = new util.ArrayList()
        //List all rows to delete
        rdd.foreach(row =&gt; {
          val delete: Delete = new Delete(row)
          delete.addColumns(Bytes.toBytes(&quot;0&quot;), Bytes.toBytes(&quot;DATA&quot;))
          deletions.add(delete)
        })
        hTable.delete(deletions)
      }
      def close(): Unit = {}
    }
  }
  </code></pre>
  <p>Any help/pointers would be greatly appreciated</p>