从Cassandra读取数据以便在Flink中处理

Question

我必须使用Flink作为流引擎处理来自Kafka的数据流。为了对数据进行分析，我需要查询Cassandra中的一些表。做这件事最好的方法是什么？我一直在Scala中寻找这样的例子。但是我找不到任何数据。如何使用Scala作为编程语言在Flink中读取来自Cassandra的数据呢？使用apache flink Java API将数据读写到cassandra中也有同样的问题。答案中提到它有多种方法。我想知道在我的情况下什么是最好的方法。而且，大多数可用的示例都是Java的。我正在寻找Scala示例。

Answer 1

我目前在Flink1.3中使用asyncIO阅读cassandra。下面是关于它的文档：

https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/asyncio.html（如果它有DatabaseClient，则使用com.datastax.drive.core.cluster）

如果您需要一个更深入的示例来使用它来阅读cassandra的内容，请告诉我，但不幸的是，我只能提供一个Java的示例。

编辑%1

下面是我使用Flink的异步I/O从Cassandra读取的代码示例。我仍在努力识别和修复一个问题，即由于某种原因（没有深入研究），单个查询返回大量数据时，异步数据流的超时被触发，尽管Cassandra似乎很好地返回了它，而且早在超时时间之前。但假设这只是我正在做的其他事情的一个bug，而不是因为这段代码，那么这对您来说应该很好（对我来说也很好地工作了几个月）：

public class GenericCassandraReader extends RichAsyncFunction<CustomInputObject, ResultSet> {

    private final Properties props;
    private Session client;

    public GenericCassandraReader(Properties props) {
        super();
        this.props = props;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        client = Cluster.builder()
                .addContactPoint(props.cassandraUrl)
                .withPort(props.cassandraPort)
                .build()
                .connect(props.cassandraKeyspace);
    }

    @Override
    public void close() throws Exception {
        client.close();
    }

    @Override
    public void asyncInvoke(final CustomInputObject customInputObject, final AsyncCollector<ResultSet> asyncCollector) throws Exception {

        String queryString = "select * from table where fieldToFilterBy='" + customInputObject.id() + "';";

        ListenableFuture<ResultSet> resultSetFuture = client.executeAsync(queryString);

        Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {

            public void onSuccess(ResultSet resultSet) {
                asyncCollector.collect(Collections.singleton(resultSet));
            }

            public void onFailure(Throwable t) {
                asyncCollector.collect(t);
            }
        });
    }
}

再次抱歉耽搁了。我希望能够解决这个错误，这样我就可以确定了，但是我想现在只有一些引用总比什么都没有要好。

编辑2

我们还意识到，由于网络延迟问题，我们最终使用了RichMapFunction，它将我们从cassandra读取的数据保持在状态。因此，该作业只需跟踪通过它的所有记录，而不必每次通过新记录时都从表中读取，以获取其中的所有记录。