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

Question

尝试使用Apache Flink从Cassandra获取数据，引用本文，我可以读取数据，但我不知道如何将其加载到DataStream对象中。代码如下：

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()));

将变量中的数据加载到数据流中

Answer 1

在Flink中创建数据流总是从ExecutionEnvironment开始。

而不是：

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

尝试：

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

然后可以使用map函数将数据类型更改为DataStream

我没有使用卡桑德拉连接器本身，所以我不知道您是否正确使用了该部件。

Answer 2

从DB读取数据-是一项有限的任务。使用CassandraInputFormat时，您应该使用DataSet API，而不是DataStream。例如：

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();
            }
        });
    }
}