如何将数据从Cassandra加载到Apache Flink DataStream

Question

尝试使用Apache Flink从Cassandra获取数据，引用此帖子，我可以读取数据，但是我没有如何将其加载到数据式对象中。以下是代码：

ClusterBuilder cb = new ClusterBuilder() {
            @Override
            public Cluster buildCluster(Cluster.Builder builder) {
                return builder.addContactPoint("localhost")
                        /*.withCredentials("hduser".trim(), "hadoop".trim())*/
                        .build();
            }
        };
CassandraInputFormat<Tuple2<UUID, String>> cassandraInputFormat = new CassandraInputFormat<Tuple2<UUID, String>>(query, cb);
cassandraInputFormat.configure(null);
cassandraInputFormat.open(null);
Tuple2<UUID, String> testOutputTuple = new Tuple2<>();
ByteArrayOutputStream res = new ByteArrayOutputStream();
res.reset();
while (!cassandraInputFormat.reachedEnd()) {
    cassandraInputFormat.nextRecord(testOutputTuple);
    res.write((testOutputTuple.f0.toString() + "," + testOutputTuple.f1).getBytes());
}
DataStream<byte[]> temp = new DataStream<byte[]>(env, new StreamTransformation<byte[]>(res.toByteArray()));

我尝试了

DataStream<byte[]> temp = new DataStream<byte[]>(env, new StreamTransformation<byte[]>(res.toByteArray()));

将数据加载到 res变量中的 DataStream<byte[]>对象中，但这不是正确的方法。我怎样才能做到这一点？我阅读Cassandra的方法是否适合流处理？

Answer 1

从db读取数据 - 是有限的任务。使用CassandRainputFormat时，您应该使用数据集API，而不是数据式。例如：

DataSet<Tuple2<Long, Date>> ds = env.createInput(executeQuery(YOUR_QUERY), TupleTypeInfo.of(new TypeHint<Tuple2<Long, Date>>() {}));
private static CassandraInputFormat<Tuple2<Long, Date>> executeQuery(String YOUR_QUERY) throws IOException {
    return new CassandraInputFormat<>(YOUR_QUERY, new ClusterBuilder() {
        private static final long serialVersionUID = 1;
            @Override
            protected Cluster buildCluster(com.datastax.driver.core.Cluster.Builder builder) {
                return builder.addContactPoints(CASSANDRA_HOST).build();
            }
        });
    }
}

Answer 2

在flink中创建数据截图始终从executionEnvironment开始。

而不是：

DataStream<byte[]> temp = new DataStream<byte[]>(env, new StreamTransformation<byte[]>(res.toByteArray())); 

尝试：

DataStream<Tuple2<UUID, String>> raw = ExecutionEnvironment.createInput(cassandraInputFormat);

您可以使用地图函数将数据类型更改为dataStream

---

我没有使用Cassandra连接器本身，所以我不知道您是否正确使用该部分。