java - Flink 从 cassandra 获取数据用作数据集时出现问题

Question

我正在尝试从 cassandra 表中提取数据用作数据集，但遇到了两个问题。

首先是 cassandraInputFormat 只返回一个元组，我不想有 tuple12 而是使用 pojo 来定义它期望返回的内容。所以我不知道这是否只是我必须接受的事情，是否有一种方法可以使用 pojo 来代替 cassandraConnector ( https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/connectors/cassandra.html )，或者如果使用 cassandraInputFormat 不是最好的方法获取数据。

另一个问题是，即使我从 cassandraInputFormat 获取的数据(无论是否是元组)我也不知道如何将其设置为数据源。对于文件、csv 和 HDFS，有很多方法 ( https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/api/java/ExecutionEnvironment.html#ExecutionEnvironment-- )，但没有一个明确适用于 cassandra。所以我的猜测是，我需要使用 cassandraInputFormat 提取数据并使用 .fromElements() 或 .fromCollecton() 之类的东西，以及正确的方法是什么。

感谢您提前提供的任何帮助!

更新:

这个“有效”(感谢 Chesnay Schepler 的帮助):

DataSet<Tuple2<String, String>> testSet = 
exEnv.createInput(cassandraInputFormat, TypeInformation.of(newTypeHint<Tuple2<String, String>>(){})); 

但是这个错误现在正在发生......

Exception in thread "main" org.apache.flink.optimizer.CompilerException: 
Error translating node 'Data Source "at execute(CodeBatchProcessorImpl.java:85) 
(org.apache.flink.batch.connectors.cassandra.CassandraInputFormat)" : NONE 
[[ GlobalProperties [partitioning=RANDOM_PARTITIONED] ]] [[ LocalProperties [ordering=null, grouped=null, unique=null] ]]':
Could not write the user code wrapper class org.apache.flink.api.common.operators.util.UserCodeObjectWrapper :
java.io.NotSerializableException: flink.streaming.code.CodeBatchProcessorImpl

进一步包括:

由以下原因引起:java.io.NotSerializedException:org.apache.flink.api.java.LocalEnvironment

更新2:

必须将环境设置为 transient 。现已修复!

Answer 1

您可以通过调用 ExecutionEnvironment#createInput(InputFormat) 来使用 CassandraInputFormat 以及与此相关的所有 InputFormat。

目前没有选项可以直接将元素读取为 POJO。最简单的解决方法是在接收器之后添加一个 MapFunction，将元组转换为您想要的 POJO。