Comatibility with spark 2.3.0 problems

[fhuertas]

This method is not valid for spark 2.3.0 because addBatch function now has a streaming dataset that cannot be stored with official batch connector

[dannybusch]

Is there already any new hint on how to write a Spark Cassandra Sink for Spark 2.3.x in Structured Streaming?

[fhuertas]

I worked on it 1 month ago and I didn't found an easy solution. I just read that there is a new spark version. I will try to check if with 2.3.1 this method is valid. If not, I should thinking other way to solve this problem.

[fhuertas]

Confirmed that Spark 2.3.1 keeps the dataframe as streaming. Then this solution is not valid.

When I have time, I will try to implement with other way, maybe using foreach

[redsk]

Hi @fhuertas , do you have any news on this?

[gitsparky]

The idea from a PR for Hadoop seems to work, see:
hortonworks-spark/shc#238

[fhuertas]

This weekend I'm going to try to adapt this code and test if it works.

 override def addBatch(batchId: Long, data: DataFrame): Unit = synchronized {
    // use a local variable to make sure the map closure doesn't capture the whole DataFrame
    val schema = data.schema
    val res = data.queryExecution.toRdd.mapPartitions { rows =>
        val converter = CatalystTypeConverters.createToScalaConverter(schema)
        rows.map(converter(_).asInstanceOf[Row])
      }

    val df = sqlContext.sparkSession.createDataFrame(res,schema)
    df.write
      .options(specifiedHBaseParams)
      .format(defaultFormat)
      .save()
  }

It looks good and could be a good solution

[fhuertas]

I have done a tests that works with 2.3.0/1/2 and it looks good, I will upload this afternoon a branch with the preliminary version.

It is the same idea that proposed in the issue: hortonworks-spark/shc#238

Thanks @gitsparky for the link