java—将flink的mapfunction与cassandra“insert”一起使用,但与“statement”一起使用时出现runtimeexception

vmjh9lq9  于 2021-06-25  发布在  Flink
关注(0)|答案(1)|浏览(370)

在与cassandra一起使用flink流进行实验时,我遇到了一个有趣的问题:在mapfunction中生成insert语句。如果我使用 DataStream<Insert> 我会弄糊涂的 RuntimeException 扔给我。但是,通过使用 DataStream<Statement> 相反,尽管我仍然使用 Insert 执行的代码中的示例。
我找到了一个解决方案(使用 DataStream<Statement> )但我仍然不清楚是什么导致了这种情况。是故意的还是窃听器?我一直在谷歌上找不到任何解释,所以不妨在这里问问有没有人知道发生了什么事。
预期输出(使用 DataStream<Statement> ):

log4j:WARN No appenders could be found for logger (org.apache.flink.api.java.typeutils.TypeExtractor).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Connected to JobManager at Actor[akka://flink/user/jobmanager_1#-638132790]
01/17/2017 15:57:42 Job execution switched to status RUNNING.
01/17/2017 15:57:42 Source: Custom Source -> Map -> Sink: Unnamed(1/1) switched to SCHEDULED 
01/17/2017 15:57:42 Source: Custom Source -> Map -> Sink: Unnamed(1/1) switched to DEPLOYING 
01/17/2017 15:57:42 Source: Custom Source -> Map -> Sink: Unnamed(1/1) switched to RUNNING 
INSERT INTO tablename (name,age) VALUES ('Test Nameson',27);
01/17/2017 15:57:42 Source: Custom Source -> Map -> Sink: Unnamed(1/1) switched to FINISHED 
01/17/2017 15:57:42 Job execution switched to status FINISHED.

错误输出(使用 DataStream<Insert> ):

Exception in thread "main" java.lang.RuntimeException: The field private java.util.List com.datastax.driver.core.querybuilder.BuiltStatement.values is already contained in the hierarchy of the class com.datastax.driver.core.querybuilder.BuiltStatement.Please use unique field names through your classes hierarchy
    at org.apache.flink.api.java.typeutils.TypeExtractor.getAllDeclaredFields(TypeExtractor.java:1762)
    at org.apache.flink.api.java.typeutils.TypeExtractor.analyzePojo(TypeExtractor.java:1683)
    at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:1580)
    at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:1479)
    at org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfoWithTypeHierarchy(TypeExtractor.java:737)
    at org.apache.flink.api.java.typeutils.TypeExtractor.privateCreateTypeInfo(TypeExtractor.java:565)
    at org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:366)
    at org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:305)
    at org.apache.flink.api.java.typeutils.TypeExtractor.getMapReturnTypes(TypeExtractor.java:120)
    at org.apache.flink.streaming.api.datastream.DataStream.map(DataStream.java:506)
    at se.hiq.bjornper.testenv.cassandra.SOCassandraQueryTest.main(SOCassandraQueryTest.java:51)

代码示例(为两种不同的情况切换注解代码):

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import com.datastax.driver.core.Statement;
import com.datastax.driver.core.querybuilder.Insert;
import com.datastax.driver.core.querybuilder.QueryBuilder;

public class SOCassandraQueryTest {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Map<String, Object>> myDataStream = env.addSource(new RichSourceFunction<Map<String, Object>>() {

            @Override
            public void run(SourceContext<Map<String, Object>> ctx) throws Exception {
                Map<String, Object> map = new HashMap<String, Object>();
                map.put("name", "Test Nameson");
                map.put("age", 27);
                ctx.collect(map);
            }

            @Override
            public void cancel() {
            }
        });

        /* Works just fine */
        DataStream<Statement> debugDatastream = myDataStream.map(new MapFunction<Map<String, Object>, Statement>() {

            @Override
            public Statement map(Map<String, Object> datarow) throws Exception {
                Insert insert = QueryBuilder.insertInto("tablename");

                for (Entry<String, Object> e : datarow.entrySet()) {
                    insert.value(e.getKey(), e.getValue());
                }
                return insert;
            }
        });

        /* Throws RuntimeException if using "Insert" instead of "Statement" */
//        DataStream<Insert> debugDatastream = myDataStream.map(new MapFunction<Map<String, Object>, Insert>() {
//
//            @Override
//            public Insert map(Map<String, Object> datarow) throws Exception {
//                Insert insert = QueryBuilder.insertInto("tablename");
//
//                for (Entry<String, Object> e : datarow.entrySet()) {
//                    insert.value(e.getKey(), e.getValue());
//                }
//                return insert;
//            }
//        });

        debugDatastream.print();

        env.execute("CassandraQueryTest");
    }
}

环境:
java 8
flink 1.1.3(来自maven包的cassabdra驱动程序)
现在有了所需的工具

byqmnocz

byqmnocz1#

flink正在分析您通过网络发送的类型,以生成快速序列化程序,并允许在构建窗口或在网络上洗牌数据时访问您的密钥。
这里的问题可能是:-当使用 Insert 作为用户类型,flink尝试生成 PojoSerializer 但是失败了 RuntimeException . 我认为这种行为是不正确的。我已经为这个问题在flink项目中提交了一份错误报告。-对于 Statement ,flink发现它不能将类型序列化为pojo,因此它将返回到泛型序列化程序(在flink的例子kryo中)。
我认为这个文档页面是我们所拥有的最接近描述flink序列化堆栈工作原理的页面:https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/types_serialization.html

相关问题