Spark Streaming无法将Kafka数据存储至HDFS问题求助

让我帮你梳理下当前遇到的问题，主要有两个核心原因，还有一些细节需要调整：

一、依赖版本不兼容（最关键的问题） #

你当前的依赖版本搭配存在明显冲突：

• Spark Streaming版本是2.2.0，但spark-streaming-kafka-0-10却用了2.0.0，Spark生态的组件必须保持版本一致，否则会出现序列化、通信层面的隐性错误，导致数据根本没被正确消费。
• kafka-clients用的是0.8.2.0，而spark-streaming-kafka-0-10是专门适配Kafka 0.10及以上版本的客户端，跨大版本的客户端会直接导致Kafka消费失败。

请替换成以下匹配的依赖：

"org.apache.kafka" % "kafka-clients" % "0.10.2.2", // 与Spark 2.2.0官方适配的Kafka客户端版本
"org.apache.spark" %% "spark-streaming" % "2.2.0",
"org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.2.0"

二、代码写入HDFS的逻辑问题 #

就算数据能被消费，你的当前代码也无法正常生成预期文件：

• saveAsTextFile接收的参数是目录名，不是文件名，Spark会自动在该目录下生成part-xxx格式的分片文件，如果你传入具体文件名，Spark会把它当作目录创建，重复运行还会因目录已存在报错。
• 没有判断RDD是否为空，若当前批次无数据，不会生成任何文件，你也无法感知。
• 直接保存ConsumerRecord对象，最终写入的是对象的toString内容，不是实际的消息值。

修改后的代码参考：

directKafkaStream.foreachRDD(rdd -> {
    // 先判断当前批次是否有数据
    if (!rdd.isEmpty()) {
        // 用时间戳生成唯一目录，避免重复创建报错
        String outputDir = "hdfs://.../sampleTest_" + System.currentTimeMillis();
        // 提取Kafka消息的value再保存，而不是整个ConsumerRecord对象
        rdd.map(ConsumerRecord::value)
           .saveAsTextFile(outputDir);
        System.out.println("成功写入 " + rdd.count() + " 条数据到HDFS：" + outputDir);
    } else {
        System.out.println("当前批次无数据可写入");
    }
    // 打印消息内容时也建议输出实际value
    rdd.foreach(record -> System.out.println("Got the record : " + record.value()));
});

三、额外检查项 #

• 先用Kafka自带的kafka-console-consumer.sh工具手动消费目标主题，确认主题内确实有数据
• 检查Spark运行用户是否有HDFS目标路径的读写权限
• 查看Spark应用的完整日志，里面通常会隐藏Kafka连接失败、权限不足等隐性报错

内容的提问来源于stack exchange，提问作者A.Rohan