如何在PyFlink中为CassandraSink配置认证凭证？

PyFlink的CassandraSink API确实没有直接暴露用户名密码的设置方法，但可以通过传递Cassandra Driver配置参数来实现认证，底层复用了Flink Java侧的Cassandra连接器能力。

具体实现步骤：

• 创建Configuration对象，添加Cassandra认证所需的配置项
• 通过set_configuration方法将配置传入Sink构建器

修改后的代码示例：

from pyflink.datastream.connectors.cassandra import CassandraSink
from pyflink.common.configuration import Configuration

# 初始化Cassandra认证配置
cassandra_conf = Configuration()
# 新版DataStax驱动的认证参数
cassandra_conf.set_string("datastax-java-driver.basic.auth.username", "your_username")
cassandra_conf.set_string("datastax-java-driver.basic.auth.password", "your_password")
# 可选：如果连接Skylla或旧版Cassandra，可指定协议版本
# cassandra_conf.set_string("datastax-java-driver.protocol.version", "V4")

cassandra_sink = CassandraSink \
        .add_sink(aggregated_stream) \
        .set_query(insert_query) \
        .set_host(CASSANDRA_HOST, int(CASSANDRA_PORT)) \
        .set_configuration(cassandra_conf)  # 注入认证配置
        .enable_ignore_null_fields() \
        .build()

cassandra_sink.set_parallelism(GLOBAL_PARALLELISM)
env.execute("Data Ingestion Job to Cassandra")

补充说明： #

1. 配置项对应关系：
   • 新版Native Protocol驱动（推荐）：使用datastax-java-driver.basic.auth.*前缀的参数
   • 旧版Thrift驱动：使用cassandra.username和cassandra.password
2. Kerberos认证场景：可以通过设置datastax-java-driver.auth.provider=kerberos，并添加Kerberos相关配置项（如datastax-java-driver.kerberos.service等）
3. 依赖检查：确保作业提交时包含对应版本的Cassandra连接器Jar包，版本需与PyFlink版本兼容

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