从火花流中的kafka消息中提取时间戳？答案

【问题标题】：Extract the time stamp from kafka messages in spark streaming?从火花流中的kafka消息中提取时间戳？
【发布时间】：2017-03-27 23:13:09
【问题描述】：

尝试从 kafka 源读取。我想从收到的消息中提取时间戳以进行结构化火花流。卡夫卡（版本 0.10.0.0）火花流（2.0.1 版）

【问题讨论】：

您能展示一下您当前代码的 sn-p 吗？
@vanekjar val ds1 = spark .readStream .format("kafka") .option("kafka.bootstrap.servers", "localhost:9092") .option("subscribe", "topicA" ) .load()

标签： apache-kafka spark-streaming spark-streaming-kafka

【解决方案1】：

spark.read
  .format("kafka")
  .option("kafka.bootstrap.servers", "your.server.com:9092")
  .option("subscribe", "your-topic")
  .load()
  .select($"timestamp", $"value")

字段“时间戳”是您要查找的内容。类型 - java.sql.Timestamp。确保您连接到 0.10 Kafka 服务器。早期版本没有时间戳。此处描述的完整字段列表 - http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#creating-a-kafka-source-for-batch-queries

【讨论】：

【解决方案2】：

我建议几件事：

假设您通过最新的Kafka Streaming Api (0.10 Kafka) 创建一个流

例如你使用依赖："org.apache.spark" %% "spark-streaming-kafka-0-10" % 2.0.1

比你创建一个流，根据上面的文档：

 val kafkaParams = Map[String, Object](
     "bootstrap.servers" -> "broker1:9092,broker2:9092",
     "key.deserializer" -> classOf[StringDeserializer],
     "value.deserializer" -> classOf[ByteArrayDeserializer],
     "group.id" -> "spark-streaming-test",
     "auto.offset.reset" -> "latest",
     "enable.auto.commit" -> (false: java.lang.Boolean))

val sparkConf = new SparkConf()
// suppose you have 60 second window
val ssc = new StreamingContext(sparkConf, Seconds(60))
ssc.checkpoint("checkpoint")

val stream = KafkaUtils.createDirectStream(ssc, PreferConsistent,
Subscribe[String, Array[Byte]](topics, kafkaParams))

您的流将是ConsumerRecord[String,Array[Byte]] 的 DStream，您可以获得时间戳和键值，如下所示：
```
stream.map { record => (record.timestamp(), record.key(), record.value())  }
```

希望对您有所帮助。

【讨论】：

如何将上面的stream.map { record => (record.timestamp(), record.key(), record.value()) 转换为DF。我是 spark 的新手，我想在转换为 DF 时也包含来自 kafka 的时间戳。你能告诉我怎么做吗
@BigD 你可以这样做：stream.foreachRDD{ (rdd: RDD[ConsumerRecord[String, Array[Byte]]], time: Time) => rdd.map(//some mapping ).toDF }
如果我做 rdd.map(//some mapping).toDF 我正在将我的值部分转换为 DF.. 我想在转换为 DF 时也包含时间戳