flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From shashank agarwal <shashank...@gmail.com>
Subject Re: Cassandra POJO sink flink 1.4.0 in scala
Date Tue, 19 Dec 2017 12:41:13 GMT
I have tried that by creating class with companion static object:

@SerialVersionUID(507L)
@Table(keyspace = "neofp", name = "order_detail")
class OrderFinal(
                       @BeanProperty var order_name: String,
                       @BeanProperty var user: String )extends Serializable
{
  def this() {
    this("NA", "NA",)
  }
}
object OrderFinal
{

}



When running with 1.4.0 it's giving following error :


java.lang.NoClassDefFoundError: Could not initialize class
com.datastax.driver.core.NettyUtil
at
com.datastax.driver.core.NettyOptions.eventLoopGroup(NettyOptions.java:96)
at com.datastax.driver.core.Connection$Factory.<init>(Connection.java:706)
at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1365)
at com.datastax.driver.core.Cluster.init(Cluster.java:162)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:333)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:308)
at com.datastax.driver.core.Cluster.connect(Cluster.java:250)
at
org.apache.flink.streaming.connectors.cassandra.CassandraSinkBase.open(CassandraSinkBase.java:88)
at
org.apache.flink.streaming.connectors.cassandra.CassandraPojoSink.open(CassandraPojoSink.java:64)
at
org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
at
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
at
org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:48)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:393)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:254)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
at java.lang.Thread.run(Thread.java:745)

12/19/2017 18:04:33 Job execution switched to status FAILING.
java.lang.NoClassDefFoundError: Could not initialize class
com.datastax.driver.core.NettyUtil
at
com.datastax.driver.core.NettyOptions.eventLoopGroup(NettyOptions.java:96)
at com.datastax.driver.core.Connection$Factory.<init>(Connection.java:706)
at com.datastax.driver.core.Cluster$Manager.init(Cluster.java:1365)
at com.datastax.driver.core.Cluster.init(Cluster.java:162)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:333)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:308)
at com.datastax.driver.core.Cluster.connect(Cluster.java:250)
at
org.apache.flink.streaming.connectors.cassandra.CassandraSinkBase.open(CassandraSinkBase.java:88)
at
org.apache.flink.streaming.connectors.cassandra.CassandraPojoSink.open(CassandraPojoSink.java:64)
at
org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:36)
at
org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:102)
at
org.apache.flink.streaming.api.operators.StreamSink.open(StreamSink.java:48)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:393)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:254)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
at java.lang.Thread.run(Thread.java:745)

On Tue, Dec 19, 2017 at 3:34 PM, shashank agarwal <shashank734@gmail.com>
wrote:

> HI,
>
> I have upgraded flink from 1.3.2 to 1.4.0. I am using cassandra sink in my
> scala application. Before sink, i was converting my scala datastream to
> java stream and sinking in Cassandra. I have created pojo class in scala
> liked that :
>
> @SerialVersionUID(507L)
> @Table(keyspace = "neofp", name = "order_detail")
> case class OrderFinal(
>                        @BeanProperty var order_name: String,
>                        @BeanProperty var user: String )extends Serializable
> {
>   def this() {
>     this("NA", "NA",)
>   }
> }
>
> and this was working fine with sink. after upgrading to 1.4.0 it's giving
> error "Query must not be null or empty."
>
> After dig into the CassandraSink code, I have found it's treating it as
> case class and running CassandraScalaProductSinkBuilder which
> do sanityCheck of query existence.
>
> So how I can create POJO class in scala so CassandraSink treats it
> as CassandraPojoSinkBuilder?
>
> For workaround now I have downgraded the only connector to 1.3.2
>
>
> Thanks
> Shashank
>
>


-- 
Thanks Regards

SHASHANK AGARWAL
 ---  Trying to mobilize the things....

Mime
View raw message