选项1：硬编码模式并在F.from_json()中使用它。

my_schema = T.StructType([
    T.StructField('a', T.IntegerType()),
    T.StructField('b', T.IntegerType()),
    T.StructField('c', T.IntegerType()),
    T.StructField('d', T.IntegerType()),
])
value = F.col('value').cast(T.StringType())
data = F.from_json(value, my_schema).alias('data')
source_df = streaming_data.select(data).select('data.*')

字符串
备选方案2：如果你想动态地推断模式，你可以使用foreachbatch。但是要注意，这是有风险的，破坏模式的改变会使流查询失败。而且不能保证模式会被正确地推断。

def parse_and_process(df: DataFrame, epoch_id: int) -> None:
    # cache the current micro batch, it will be scanned more than once
    df.persist()
    # infer the schema of the current batch
    spark = SparkSession.getActiveSession()
    value = F.col('value').cast(T.StringType())
    inferred_df = spark.read.json(
        df.select(value).rdd.map(lambda x: x[0]),
        dropFieldIfAllNull=True
    )
    inferred_schema = inferred_df.schema
    # parse the json with the schema
    res_df = df.withColumn('data', F.from_json(value, inferred_schema))
    # process the DataFramee, it's not a streaming DataFrame anymore.
    res_df.write....
    df.unpersist()
streaming_data.writeStream.foreachBatch(parse_and_process).start()

型