我想在foreachrdd方法中检索rdd的每条记录上的每个kafka偏移量。我的主题有一个分区，所以我的rdd也有一个分区。我基本上会这样做：

dStream.foreachRDD { rdd =>
  if (!rdd.isEmpty) {
    //get offset first value of the offset
    val firstOffset = rdd.asInstanceOf[HasOffsetRanges].offsetRanges(0).fromOffset
    val rddWithOffset = rdd.map(_.value)
      .zipWithIndex()
      .map{ case (v,i) => (v,i + firstOffset)}
  }
}

例如，在我的producer中，我使用循环发送消息，并将索引放置在名为position的列中，如下所示：

+------+-----+--------+
|  name|  age|position|
+------+-----+--------+
|johnny|   26|       1|
| chloe|   42|       2|
| brian|   19|       3|
| eliot|   35|       4|
+------+-----+--------+

不幸的是，我注意到，在我的消费者中添加offset列时，订单没有得到维护：

+------+-----+--------+------+
|  name|  age|position|offset|
+------+-----+--------+------+
|johnny|   26|       1|     1|
| chloe|   42|       2|     3|
| brian|   19|       3|     4|
| eliot|   35|       4|     2|
+------+-----+--------+------+

我好像在这个过程中失去了秩序。你知道吗？谢谢
顺便说一下，我的java制作人是这样的：

KafkaRestProducer<String, Object> producer = new KafkaRestProducer<>(props);

ArrayList<String> names = new ArrayList<String>()
names.add("johnny")
names.add("chloe")
names.add("brian")
names.add("eliot")

ArrayList<Integer> ages = ArrayList<Integer>()
names.add(26)
names.add(42)
names.add(19)
names.add(35)

for (int i = 0; i < 3; ++i) {

    String name = names(i)
    Int age = ages(i)     
    Person person = Person
        .newBuilder()
        .setName(name)
        .setAge(age)
        .setPosition(i)
        .build();

    ProducerRecord<String, Object> record = new ProducerRecord<>("/apps/PERSON/streams:myTopic", name, person);

    producer.send(record, null);
    System.out.println(i);
}