cassandra 移除重复项而不随机播放Spark

kokeuurv  于 2022-11-05  发布在  Cassandra
关注(0)|答案(1)|浏览(110)

我有一个Cassandra表XYX,包含列(id uuid,插入时间戳,标题文本)
其中id和insert是组合主键。
我正在使用Dataframe,在我的spark shell中,我正在获取id和标题列。我希望基于id和标题列有不同的行。
我看到很多洗牌,这不是这样,因为SparkCassandra连接器确保所有行为一个给定的Cassandra分区是在同一Spark分区。
在提取之后,我使用dropDuplicates来获取不同的记录。

zpgglvta

zpgglvta1#

Spark Dataframe API还不支持自定义分区器。因此连接器无法将C* 分区器引入Dataframe引擎。RDD Spark API支持来自其他方面的自定义分区器。因此,您可以将数据加载到RDD中,然后将其转换为df。以下是有关C* 分区器用法的连接器文档:https://github.com/datastax/spark-cassandra-connector/blob/master/doc/16_partitioning.md
keyBy()函数允许您定义用于分组键列
这是一个工作示例。它不短,所以我希望有人可以改进它:

//load data into RDD and define a group key
val rdd = sc.cassandraTable[(String, String)] ("test", "test")
   .select("id" as "_1", "header" as "_2")
   .keyBy[Tuple1[Int]]("id")
// check that partitioner is CassandraPartitioner
rdd.partitioner
// call distinct for each group, flat it, get two column DF
val df = rdd.groupByKey.flatMap {case (key,group) => group.toSeq.distinct}
    .toDF("id", "header")

相关问题