flink cassandraoutputformat元组需要冻结值

k4ymrczo  于 2021-06-25  发布在  Flink
关注(0)|答案(2)|浏览(305)

我有一个flink项目,它将作为批处理作业在cassandra表中插入数据。我已经有了一个flink stream项目,它正在将pojo写入同一个cassandra表,但是cassandraoutputformat需要数据作为元组(希望在某个时候可以像cassandrasink那样接受pojo)。这就是我的观点:

@Table(keyspace="mykeyspace", name="mytablename")
public class AlphaGroupingObject implements Serializable {

    @Column(name = "jobId")
    private String jobId;
    @Column(name = "datalist")
    @Frozen("list<frozen<dataobj>")
    private List<CustomDataObj> dataobjs;
    @Column(name = "userid")
    private String userid;

    //Getters and Setters
}

我用这个pojo制作的元组数据集:

DataSet<Tuple3<String, List<CustomDataObj>, String>> outputDataSet = listOfAlphaGroupingObject.map(new AlphaGroupingObjectToTuple3Mapper());

下面是触发输出的行:

outputDataSet.output(new CassandraOutputFormat<>("INSERT INTO mykeyspace.mytablename (jobid, datalist, userid) VALUES (?,?,?);", clusterThatWasBuilt));

现在我遇到的问题是,当我试图运行它时,当它试图将它输出到cassandra表时,我得到了这个错误:

Caused by: com.datastax.driver.core.exceptions.CodecNotFoundException: 
Codec not found for requested operation: [frozen<mykeyspace.dataobj> <-> flink.custom.data.CustomDataObj]

所以我知道当它是pojo时,我只需要将@freezed注解添加到字段中,但我不知道如何为元组添加注解。解决这个问题的最佳/正确方法是什么?或者我做了一些不必要的事情,因为实际上有一种方法可以通过cassandraoutputformat发送pojo,我只是没有找到?
谢谢你的帮助!
编辑:
下面是customdataobj类的代码:

@UDT(name="dataobj", keyspace = "mykeyspace")
public class CustomDataObj implements Serializable {

    @Field(name = "userid")
    private String userId;

    @Field(name = "groupid")
    private String groupId;

    @Field(name = "valuetext")
    private String valueText;

    @Field(name = "comments")
    private String comments;

    //Getters and setters
}

编辑2
包括customdataobj绑定到的cassandra中的表模式和mytablename模式。

CREATE TYPE mykeyspace.dataobj (
    userid text,
    groupid text,
    valuetext text,
    comments text
);

CREATE TABLE mykeyspace.mytablename (
    jobid text,
    datalist list<frozen<dataobj>>,
    userid text,
    PRIMARY KEY (jobid, userid)
);
daupos2t

daupos2t1#

添加 UDT 上的注解 CustomDataObj

@UDT(name = "dataobj")
public class CustomDataObj { 
    //...... 
}

编辑
改变 jobid 注解到 @Column(name = "jobid") 以及 dataobjs 冻结注解到 @Frozen ```
@Table(keyspace="mykeyspace", name="mytablename")
public class AlphaGroupingObject implements Serializable {

@Column(name = "jobid")
private String jobId;

@Column(name = "datalist")
@Frozen
private List<CustomDataObj> dataobjs;
@Column(name = "userid")
private String userid;

//Getters and Setters

}

f4t66c6m

f4t66c6m2#

我相信我已经找到了一个更好的方法,而不是必须为cassandraoutputformat提供一个元组,但从技术上讲,它仍然不能回答这个问题,所以我不会将此标记为答案。最后我使用了cassandra的对象Map器,这样就可以将pojo发送到表中。仍然需要验证数据是否成功地到达表中,以及所有内容是否按照实现方式正常工作,但我觉得这将帮助任何面临类似问题的人。
以下是概述解决方案的文档:http://docs.datastax.com/en/developer/java-driver/2.1/manual/object_mapper/using/

相关问题