使用apache flink Java API将数据读写到cassandra中

Question

我打算使用apache flink将数据读/写到使用flink的cassandra中。我希望使用flink-connector-cassandra，但我没有找到关于连接器的好文档/示例。

你能给我指出正确的方式来读写数据从cassandra使用Apache Flink。我只看到了纯粹用于写作的sink示例？apache flink也是用来从cassandra读取数据的，是否与apache spark类似？

Answer 1

我也有同样的问题，这就是我要找的。我不知道这是不是太简单了，以满足你的需要，但我想我还是应该展示它。

ClusterBuilder cb = new ClusterBuilder() {
        @Override
        public Cluster buildCluster(Cluster.Builder builder) {
            return builder.addContactPoint("urlToUse.com").withPort(9042).build();
        }
    };

    CassandraInputFormat<Tuple2<String, String>> cassandraInputFormat = new CassandraInputFormat<>("SELECT * FROM example.cassandraconnectorexample", cb);

    cassandraInputFormat.configure(null);
    cassandraInputFormat.open(null);

    Tuple2<String, String> testOutputTuple = new Tuple2<>();
    cassandraInputFormat.nextRecord(testOutputTuple);

    System.out.println("column1: " + testOutputTuple.f0);
    System.out.println("column2: " + testOutputTuple.f1);

我找到了“CassandRainPutFormat”类的代码并了解了它的工作原理（http://www.javatips.net/api/flink-master/flink-connectors/flink-connectors/cassandra/src/main/java/org/apache/flink/batch/connectors/cassandrainPutFormat.java）。我真的希望它只是一个格式，而不是完整的类阅读从卡桑德拉的名字，我有一种感觉其他人可能也在想同样的事情。