Pyspark读取Kafka DirectStream转DataFrame写入PostgreSQL问题咨询

• 你当前执行x = lines.take(1)报错的直接原因：kafkaStream.map(lambda x: json.loads(x[1])).pprint()的pprint()方法无返回值，你赋值给lines的变量实际为None，自然无法调用take方法。
• 你使用的是Spark早期的DStream API，目前官方更推荐使用Structured Streaming处理流数据，对接Kafka、转DataFrame、写入外部存储的逻辑更简洁统一。

如果要继续使用现有DStream逻辑，可以通过foreachRDD处理每个批次的流数据，转换为DataFrame：

import json
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from pyspark.sql import SparkSession

# 定义单例SparkSession，避免重复创建
def get_spark_session_instance(sparkConf):
    if "sparkSessionSingletonInstance" not in globals():
        globals()["sparkSessionSingletonInstance"] = SparkSession \
            .builder \
            .config(conf=sparkConf) \
            .getOrCreate()
    return globals()["sparkSessionSingletonInstance"]

# 原有初始化逻辑
cols = ['id','name']
para_seconds = 10
sc = SparkContext(appName="PythonSparkStreamingKafka_Testing_01")
sc.setLogLevel("WARN")
ssc = StreamingContext(sc, para_seconds)
brokers = 'localhost:9092'
topic = "testing_topic"
kafkaStream = KafkaUtils.createDirectStream(ssc, [topic],{'bootstrap.servers':brokers})
# 去掉末尾的pprint()，保留map后的DStream
lines = kafkaStream.map(lambda x: json.loads(x[1]))

# 处理每个批次的RDD，转换为DataFrame并写入PG
def process_rdd(time, rdd):
    if rdd.isEmpty():
        return
    spark = get_spark_session_instance(rdd.context.getConf())
    # 适配嵌套JSON结构提取字段，如需全量JSON字符串可改为 row_rdd = rdd.map(lambda x: (json.dumps(x),))，schema对应改为['raw_json_str']
    row_rdd = rdd.map(lambda x: (x["id"]["testing"], x["name"]["testing"]))
    df = spark.createDataFrame(row_rdd, schema=cols)
    # 写入PostgreSQL逻辑
    df.write \
      .format("jdbc") \
      .option("url", "jdbc:postgresql://你的PG地址:端口/库名") \
      .option("dbtable", "目标表名") \
      .option("user", "PG用户名") \
      .option("password", "PG密码") \
      .option("driver", "org.postgresql.Driver") \
      .mode("append") \
      .save()
    # 测试时可打印查看DF内容
    df.show()

lines.foreachRDD(process_rdd)

ssc.start()
ssc.awaitTermination()

官方主推的流处理API，无需手动管理SparkSession实例，代码逻辑更简洁：

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType

spark = SparkSession \
    .builder \
    .appName("StructuredKafkaToPG") \
    .getOrCreate()
spark.sparkContext.setLogLevel("WARN")

# 定义Kafka消息的JSON schema，适配你的嵌套结构
json_schema = StructType([
    StructField("id", StructType([StructField("testing", StringType())])),
    StructField("name", StructType([StructField("testing", StringType())]))
])

# 读取Kafka流
df = spark \
  .readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "localhost:9092") \
  .option("subscribe", "testing_topic") \
  .load()

# 解析value为JSON，提取字段得到目标DataFrame
value_df = df.select(from_json(col("value").cast("string"), json_schema).alias("data")) \
             .select(col("data.id.testing").alias("id"), col("data.name.testing").alias("name"))

# 如需全量JSON字符串格式的DataFrame，用这行代替上面的解析逻辑：
# value_df = df.select(col("value").cast("string").alias("raw_json_str"))

# 批量写入PostgreSQL逻辑
def write_to_pg(batch_df, batch_id):
    batch_df.write \
      .format("jdbc") \
      .option("url", "jdbc:postgresql://你的PG地址:端口/库名") \
      .option("dbtable", "目标表名") \
      .option("user", "PG用户名") \
      .option("password", "PG密码") \
      .option("driver", "org.postgresql.Driver") \
      .mode("append") \
      .save()

query = value_df.writeStream \
    .foreachBatch(write_to_pg) \
    .trigger(processingTime='10 seconds') \
    .start()

query.awaitTermination()

注意 #

两种方案提交任务时，都需要携带对应版本的Kafka依赖、PostgreSQL JDBC驱动包。

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