gpt4 book ai didi

scala - 将自定义编解码器添加到 CassandraConnector

转载 作者:行者123 更新时间:2023-12-01 10:32:56 24 4
gpt4 key购买 nike

有没有办法在 CassandraConnector 实例化时注册自定义编解码器?

我目前每次调用 cassandraConnector.withSessionDo 时都在注册我的编解码器

val cassandraConnector = CassandraConnector(ssc.sparkContext.getConf)
...
...
.mapPartitions(partition => {
cassandraConnector.withSessionDo(session => {
// register custom codecs once for each partition so it isn't loaded as often for each data point
if (partition.nonEmpty) {
session.getCluster.getConfiguration.getCodecRegistry
.register(new TimestampLongCodec)
.register(new SummaryStatsBlobCodec)
.register(new JavaHistogramBlobCodec)
}

这样做似乎有点像反模式。它也确实堵塞了我们的日志,因为我们有一个每 30 秒运行一次的 spark 流服务,它正在用以下内容填充我们的日志:

16/11/01 14:14:44 WARN CodecRegistry: Ignoring codec SummaryStatsBlobCodec [blob <-> SummaryStats] because it collides with previously registered codec SummaryStatsBlobCodec [blob <-> SummaryStats]
16/11/01 14:14:44 WARN CodecRegistry: Ignoring codec JavaHistogramBlobCodec [blob <-> Histogram] because it collides with previously registered codec JavaHistogramBlobCodec [blob <-> Histogram]
16/11/01 14:14:44 WARN CodecRegistry: Ignoring codec TimestampLongCodec [timestamp <-> java.lang.Long] because it collides with previously registered codec TimestampLongCodec [timestamp <-> java.lang.Long]

编辑:

我试过像这样立即注册它们:

val cassandraConnector = CassandraConnector(ssc.sparkContext.getConf)
cassandraConnector.withClusterDo(cluster => {
cluster.getConfiguration.getCodecRegistry
.register(new TimestampLongCodec)
.register(new SummaryStatsBlobCodec)
.register(new JavaHistogramBlobCodec)
})

这 ^ 在本地工作,但是当部署到我们的 mesos 集群时,它找不到编解码器。我假设这是因为它只在驱动程序中本地注册那些,并且从不将它们添加到执行程序版本中。

最佳答案

更好的方法是覆盖 cassandra 连接工厂,像这样

import com.datastax.driver.core.Cluster
import com.datastax.spark.connector.cql.{CassandraConnectionFactory, CassandraConnectorConf, DefaultConnectionFactory}
object MyConnectionFactory extends CassandraConnectionFactory {
override def createCluster(conf: CassandraConnectorConf): Cluster = {
val cluster = DefaultConnectionFactory.createCluster(conf)
cluster.getConfiguration.getCodecRegistry
.register(new TimestampLongCodec)
.register(new SummaryStatsBlobCodec)
.register(new JavaHistogramBlobCodec)
cluster
}
}

并设置spark.cassandra.connection.factory参数指向类

关于scala - 将自定义编解码器添加到 CassandraConnector,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/40363611/

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com