在Apache·Flink，通过 CassandraSink 在cassandra中插入一行。但我找不到删除行的方法。
我也试着写自定义Flume，但我得到了 NotSerializableException . 如何构造删除操作的代码？

public class MyCassandraSink implements SinkFunction<String> {

    private Cluster cluster = Cluster.builder()
            .addContactPoint("127.0.0.1")
            .build();

    private Session cassandra = cluster.connect("mykeyspace");

    @Override
    public void invoke(String value, Context context) throws Exception {
        cassandra.execute("SOME DELETE QUERY");
    }
}

Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: [com.datastax.driver.core.SessionManager@3b0fe47a] is not serializable. The object probably contains or references non serializable fields.
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:151)
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:126)
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:71)
    at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.clean(StreamExecutionEnvironment.java:1574)
    at org.apache.flink.streaming.api.datastream.DataStream.clean(DataStream.java:185)
    at org.apache.flink.streaming.api.datastream.DataStream.addSink(DataStream.java:1227)
    at com.meshkan.streaming.entry.EventListener.main(EventListener.java:42)
Caused by: java.io.NotSerializableException: com.datastax.driver.core.SessionManager
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at java.util.concurrent.CopyOnWriteArrayList.writeObject(CopyOnWriteArrayList.java:973)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1140)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:586)
    at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:133)
    ... 9 more

2条答案

按热度按时间

xggvc2p61#

我找到了解决办法，但我不喜欢。cassandrapojoinputformat可用于删除和更新行(我还将其用于select，顾名思义，这似乎就是它的本意。）
事实上，这是唯一的救赎美德。我一直在用这个直到我找到一个优雅的解决方案。我还在找。。。

CassandraPojoInputFormat<MyThingyConnector> myThingyCassandraPojoInputFormat =
new CassandraPojoInputFormat<MyThingyConnector>(
"DELETE FROM " + dbKeyspace + ".<table_name> <where clause>",
clusterBuilder,
MyThingyConnector.class);

myThingyCassandraPojoInputFormat.configure(null);
myThingyCassandraPojoInputFormat.open(cassandraInputSplit);
myThingyCassandraPojoInputFormat.close();

赞(0）回复(0）举报 2021-06-24

nxagd54h2#

要实现自己的insert vs delete逻辑，请创建一个扩展 CassandraSinkBase ，并实施 send() 方法。看到了吗 AbstractCassandraTupleSink 举个例子。注意如何 CassandraSinkBase 通过使cassandra客户机暂时化并在 open() 打电话。

如何使用ApacheFlink删除cassandra中的一行？

2条答案

相关问题

热门标签

最新问答