写入 Amazon Keyspaces 时出现 CodecNotFoundException

Question

我正在尝试将 Spark DF 写入 AWS Keyspaces。随机一些记录得到更新，一些记录抛出这个异常

com.datastax.oss.driver.api.core.type.codec.CodecNotFoundException: Codec not found for requested operation: [INT <-> java.lang.String]
at com.datastax.oss.driver.internal.core.type.codec.registry.CachingCodecRegistry.createCodec(CachingCodecRegistry.java:609)
at com.datastax.oss.driver.internal.core.type.codec.registry.DefaultCodecRegistry.load(DefaultCodecRegistry.java:95)
at com.datastax.oss.driver.internal.core.type.codec.registry.DefaultCodecRegistry.load(DefaultCodecRegistry.java:92)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3527)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2276)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2154)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache$Segment.get(LocalCache.java:2044)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache.get(LocalCache.java:3951)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache.getOrLoad(LocalCache.java:3973)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4957)
at com.datastax.oss.driver.shaded.guava.common.cache.LocalCache$LocalLoadingCache.getUnchecked(LocalCache.java:4963)
at com.datastax.oss.driver.internal.core.type.codec.registry.DefaultCodecRegistry.getCachedCodec(DefaultCodecRegistry.java:117)
at com.datastax.oss.driver.internal.core.type.codec.registry.CachingCodecRegistry.codecFor(CachingCodecRegistry.java:258)
at com.datastax.oss.driver.internal.core.data.ValuesHelper.encodePreparedValues(ValuesHelper.java:112)
at com.datastax.oss.driver.internal.core.cql.DefaultPreparedStatement.bind(DefaultPreparedStatement.java:158)

我的键空间table架构是

CREATE TABLE test_ks.test_table_ttl (
    consumer_id TEXT PRIMARY KEY,
    ttl_col map<text, frozen<tuple<text, text>>>
);

抛出错误的代码块是这个。

val rowKey =   // some string
val mapKey =   // some string
val mapValue = mapValueTupleType.newValue(tuple_value)
val mapData = ImmutableMap.builder().put(mapKey, mapValue).build()
batch.addStatement(prep_statement.bind(mapData, rowKey)) // <--- error on this line

Answer 1

尝试转换为 RDD 然后写入。尝试将数据存储为 json blob，而不是嵌套集合。

val myRdd = myDataframe.rdd

implicit val c = connectToKeyspaces

myRdd.saveToCassandra("aws_sample_keyspace","events_tables")

Answer 2

目前，AWS Keyspaces 不允许冻结类型。这是 Keyspaces 中的一个错误，允许使用冻结类型创建 table，但它在插入期间抛出异常。

唯一可能的方法是按照@MikeJPR 的建议使用JSON 存储数据。

写入 Amazon Keyspaces 时出现 CodecNotFoundException

CodecNotFoundException while writing to Amazon Keyspaces

java

scala

datastax-java-driver

datastax

amazon-keyspaces