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

Question

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

您能否指出使用 Apache Flink 从 cassandra 读取和写入数据的正确方法。我只看到纯粹用于写入的接收器示例？ apache flink 是否也用于从类似于 apache spark 的 cassandra 读取数据？

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-connector-cassandra/src/main/java/org/apache/flink/batch/connectors/cassandra/CassandraInputFormat.java )。老实说，我希望它只是一种格式，而不是基于名称的 Cassandra 的完整阅读类(class)，我觉得其他人可能会这么想。