如何在PySpark DataFrame中使用CQL特定函数实现数据过滤？

针对你遇到的PySpark DataFrame不支持Cassandra专属CQL函数（如maxTimeuuid()/minTimeuuid()）的问题，以下是两种实用的解决方法：

方案1：直接执行原生CQL查询读取数据 #

PySpark Cassandra连接器支持直接传入完整的CQL语句加载数据，完全复用你在cqlsh中验证过的过滤逻辑，避开Spark对CQL函数的解析限制。

示例代码：

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("CassandraDataMigration") \
    .config("spark.cassandra.connection.host", "你的Cassandra主机地址") \
    .config("spark.cassandra.connection.port", "9042") \
    .getOrCreate()

# 定义包含Cassandra专属函数的原生CQL
cql_query = """
SELECT * FROM ns.table 
WHERE pk1 = 'val1' 
  AND pk2 = 'val2' 
  AND timestamp > maxTimeuuid('2023-01-01 00:00:00') 
  AND timestamp < minTimeuuid('2023-12-31 23:59:59')
"""

# 加载查询结果为DataFrame
df = spark.read \
    .format("org.apache.spark.sql.cassandra") \
    .option("query", cql_query) \
    .load()

# 迁移到目标Cassandra集群
df.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="table", keyspace="目标命名空间") \
    .mode("append") \
    .save()

方案2：提前计算Timeuuid范围值，用常规过滤 #

如果偏好使用DataFrame链式操作，可以在Python端提前计算出maxTimeuuid()/minTimeuuid()对应的实际UUID值，再用常规的字段比较语法过滤。

示例代码：

from pyspark.sql import SparkSession
import uuid
from datetime import datetime

# 自定义函数：将datetime转换为Cassandra的minTimeuuid/maxTimeuuid对应值
def datetime_to_min_timeuuid(dt):
    timestamp = int(dt.timestamp() * 1000)
    return uuid.UUID(int=timestamp << 64)

def datetime_to_max_timeuuid(dt):
    timestamp = int(dt.timestamp() * 1000)
    return uuid.UUID(int=timestamp << 64 | 0xFFFFFFFFFFFFFFFF)

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("CassandraDataMigration") \
    .config("spark.cassandra.connection.host", "你的Cassandra主机地址") \
    .getOrCreate()

# 定义时间范围并转换为对应UUID
start_dt = datetime(2023, 1, 1, 0, 0, 0)
end_dt = datetime(2023, 12, 31, 23, 59, 59)
min_uuid = datetime_to_min_timeuuid(start_dt)
max_uuid = datetime_to_max_timeuuid(end_dt)

# 读取数据并过滤
df = spark.read \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="table", keyspace="ns") \
    .load() \
    .filter(
        (df.pk1 == 'val1') & 
        (df.pk2 == 'val2') & 
        (df.timestamp > min_uuid) & 
        (df.timestamp < max_uuid)
    )

# 写入目标集群
df.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="table", keyspace="目标命名空间") \
    .mode("append") \
    .save()

关于token()函数的补充处理 #

如果需要使用token()函数，同样可以用方案1的原生CQL方式直接写，或者在Python端根据Cassandra分区器计算token值（比如Murmur3分区器可以用mmh3库计算哈希值），再用常规过滤语法。

内容的提问来源于stack exchange，提问作者Yuri Tarasyuk