如何使用PySpark为Kafka Streaming创建DataFrame（环境：Kafka2.11-1.0.1、Spark2.0.2）

嘿，针对你用Kafka 2.11-1.0.1和Spark 2.0.2创建Kafka流的DataFrame需求，我整理了具体的实现步骤，咱们一步步来：

Spark 2.0.2需要搭配对应版本的Kafka连接器，确保你的项目里引入了正确的依赖：

Maven依赖 #

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
    <version>2.0.2</version>
</dependency>
<!-- 如果用传统Spark Streaming的话，还要加这个 -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
    <version>2.0.2</version>
</dependency>

SBT依赖 #

libraryDependencies += "org.apache.spark" %% "spark-sql-kafka-0-10" % "2.0.2"
// 传统Streaming的依赖
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.0.2"

Spark 2.0.2已经支持Structured Streaming，这种方式可以直接从Kafka读取流数据并生成DataFrame，是最直接的方式：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

// 初始化SparkSession
val spark = SparkSession.builder()
  .appName("KafkaToDataFrame")
  .master("local[*]") // 生产环境去掉这个，用集群模式
  .getOrCreate()

// 读取Kafka流数据，生成DataFrame
val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-broker:9092") // 替换成你的Kafka地址
  .option("subscribe", "your-topic-name") // 替换成要订阅的topic
  .option("startingOffsets", "latest") // 可以选earliest/latest或者指定偏移量
  .load()

// Kafka原生的DataFrame包含这些字段：key, value, topic, partition, offset, timestamp, timestampType
// 如果你需要解析value（比如JSON格式），可以做如下转换
import spark.implicits._
val parsedDF = kafkaDF.selectExpr("CAST(value AS STRING)")
  .as[String]
  .map(json => {
    // 这里写你的JSON解析逻辑，比如转换成case class
    // 示例：假设JSON是{"id":1,"name":"test"}，对应Case Class User
    // User(json.split(",")(0).split(":")(1).toInt, json.split(",")(1).split(":")(1).replace("\"",""))
  })
  .toDF()

// 输出结果（比如控制台输出，生产环境可以写去HDFS/Kafka等）
val query = parsedDF.writeStream
  .outputMode("append")
  .format("console")
  .start()

query.awaitTermination()

如果你还在使用传统的Spark Streaming（DStream），可以把DStream转换成DataFrame来处理：

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{StreamingContext, Seconds}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

// 初始化StreamingContext
val conf = new SparkConf().setAppName("KafkaDStreamToDataFrame").setMaster("local[*]")
val ssc = new StreamingContext(conf, Seconds(5)) // 5秒一个批次

// 配置Kafka参数
val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "your-kafka-broker:9092",
  "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "group.id" -> "spark-consumer-group",
  "auto.offset.reset" -> "latest",
  "enable.auto.commit" -> (false: java.lang.Boolean)
)

val topics = Array("your-topic-name")
val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

// 把DStream转换成DataFrame
stream.foreachRDD { rdd =>
  val spark = SparkSession.builder().config(rdd.sparkContext.getConf).getOrCreate()
  import spark.implicits._

  // 提取Kafka的value字段，生成DataFrame
  val df = rdd.map(record => record.value()).toDF("kafka_value")
  
  // 这里可以对DataFrame进行各种操作，比如SQL查询
  df.createOrReplaceTempView("kafka_data")
  val resultDF = spark.sql("SELECT * FROM kafka_data WHERE kafka_value LIKE '%keyword%'")
  
  // 输出结果
  resultDF.show()
}

ssc.start()
ssc.awaitTermination()

• 替换代码中的your-kafka-broker:9092和your-topic-name为你实际的Kafka地址和topic名称
• startingOffsets和auto.offset.reset参数根据你的需求选择，earliest会从最开始的偏移量读取，latest从最新的开始
• 生产环境不要用master("local[*]")，要提交到Spark集群运行

内容的提问来源于stack exchange，提问作者Rj1