scala – 从Spark中的cassandra表中删除

发布时间：2020-12-16 09:54:08 所属栏目：安全来源：网络整理

导读：我正在使用Spark和cassandra.我正在从我的表中读取一些行,以便使用PrimaryKey删除主题.这是我的代码： val lines = sc.cassandraTable[(String,String,String)](CASSANDRA_SCHEMA,table). select("a","b","c","d"). where("d=?",d).cache()lines.foreach(r =

我正在使用Spark和cassandra.我正在从我的表中读取一些行,以便使用PrimaryKey删除主题.这是我的代码：

val lines = sc.cassandraTable[(String,String,String)](CASSANDRA_SCHEMA,table).
  select("a","b","c","d").
  where("d=?",d).cache()

lines.foreach(r => {
    val session: Session = connector.openSession
    val delete = s"DELETE FROM "+CASSANDRA_SCHEMA+"."+table+" where channel='"+r._1 +"' and ctid='"+r._2+"'and cvid='"+r._3+"';"
    session.execute(delete)
    session.close()
})

但是这种方法为每一行创建一个会话,这需要很多时间.那么是否可以使用sc.CassandraTable或其他更好的方法删除我的行.

谢谢

解决方法

我不认为Cassandra Connector目前支持删除.要分摊连接设置的成本,建议的方法是将操作应用于每个分区.

所以你的代码看起来像这样：

lines.foreachPartition(partition => {
    val session: Session = connector.openSession //once per partition
    partition.foreach{elem => 
        val delete = s"DELETE FROM "+CASSANDRA_SCHEMA+"."+table+" where     channel='"+elem._1 +"' and ctid='"+elem._2+"'and cvid='"+elem._3+"';"
        session.execute(delete)
    }
    session.close()
})

您还可以查看使用DELETE FROM … WHERE pk IN(列表)并使用类似的方法为每个分区构建列表.这将是更高性能,但可能会破坏非常大的分区,因为列表将变得相当长.在应用此功能之前重新分区目标RDD将有所帮助.

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!