有几件事我觉得你做错了。
你需要通过考试 Map[TopicAndPartition, Long] ，而当前您有 Tuple2[TopicAndPartition, Long] . 所以你需要：

val fromOffsets: Map[TopicAndPartition, Long] = 
    Map(TopicAndPartition(metrics_rs.getString(1), 
                          metrics_rs.getInt(2)) -> metrics_rs.getLong(3))

你说你的报税表是从哪来的 createDirectStream 是类型的元组 (String, String) ，但你的 messageHandler 值是一个 Int . 如果要返回包含键值对的元组，则需要：

val messageHandler: MessageAndMetadata[String, String] => (String, String) =
  (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())

修复后，应编译：

val stream = KafkaUtils
              .createDirectStream[String, String,
                      StringDecoder, StringDecoder,
                      (String, String)] (ssc, 
                                         kafkaParams, 
                                         fromOffsets, 
                                         messageHandler)