Spark Session无法向Cassandra插入数据，请求技术排查

以下是针对你遇到的问题的具体排查方向：

1. 确认写入逻辑是否实际执行 #

• 检查Airflow DAG调用的脚本：确保触发DAG时，执行的脚本中写入Cassandra的流处理代码已经取消注释。Airflow可能存在脚本同步延迟（比如Git拉取不及时、Worker节点本地文件未更新），直接到Worker节点查看实际运行的脚本内容。
• 查看Spark作业日志：通过Spark UI（默认http://<spark-master-ip>:8080）找到对应作业，查看Driver和Executor日志，确认是否有写入Cassandra相关的日志输出，或者是否存在未捕获的异常。

2. 验证Kafka数据流是否正常 #

• 检查Kafka主题数据：执行命令：

  kafka-console-consumer.sh --bootstrap-server <kafka-broker-ip>:9092 --topic <your-target-topic> --from-beginning
  

  确认主题中是否有数据流入。如果Kafka无数据，Spark流处理自然没有数据可写入。
• 核对Spark的Kafka配置：检查Spark作业中bootstrap.servers、主题名称是否与实际一致，起始偏移量设置是否合理（比如设为latest但主题无新数据时，会读取不到数据）。

3. 检查Spark与Cassandra的连接及权限 #

• 确认Cassandra连接参数：检查Spark作业中Cassandra的contactPoints、端口（默认9042）、keyspace（spark_streams）、表名（created_users）是否配置正确。参数错误可能导致写入失败但无明显报错。
• 验证权限：执行Spark作业的用户是否拥有spark_streams.created_users表的INSERT权限。用CQL命令检查：

  LIST PERMISSIONS ON spark_streams.created_users TO '<spark-exec-user>';
  

  若无权限，执行授权：

  GRANT INSERT ON spark_streams.created_users TO '<spark-exec-user>';
  

4. 检查Airflow DAG的执行逻辑 #

• 确认任务依赖：确保Spark流处理任务在Kafka生产者任务之后执行。如果任务顺序颠倒，Spark启动时Kafka还未产生数据，且流处理未配置持续监听，就会无数据写入。
• 核对执行模式：如果Spark流处理是一次性批处理模式，需确保任务执行时Kafka已有数据；若要处理实时数据，需配置为持续流处理模式。

5. 验证数据结构与表结构匹配性 #

• 对比字段结构：检查Spark输出的数据字段名称、数据类型是否与Cassandra表完全一致。类型不匹配（如Spark输出String但Cassandra表为Int）可能导致静默写入失败，需查看Spark日志中的错误信息。
• 检查主键配置：写入的数据是否包含完整的主键字段，且主键字段不为空。Cassandra会拒绝缺少主键或主键为空的写入请求，且默认不返回错误（需开启严格模式才能捕获）。

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