KafkaUtils API |偏移管理|火花流

Question

我正在尝试为仅一次语义管理 kafka 偏移量。

使用偏移图创建直接流时面临的问题如下：

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

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

这里，

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

还有

metrics_rs = metricsStatement.executeQuery("SELECT part,off from metrics.txn_offsets where topic='"+t+''' )

我想我在声明风格上做错了......如果你能帮忙的话。 编译错误说“createDirectStream 的类型参数太多”

Answer 1

我发现你做错了几件事。

您需要传递一个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)