事件中心没有输出

Question

我正在尝试从事件中心读取数据，但结果它只返回空值。

我将一个数据帧转换为json以发送到eventhub

dfSource_pandas = df.toPandas()
type(dfSource_pandas)
payload = dfSource_pandas.to_json(orient="records")
print(payload)

这是数据的模式

root
|-- alertId: string (nullable = true)
|-- score: double (nullable = true)
|-- severity: double (nullable = true)

我正在尝试使用从eventhub读取数据

from pyspark.sql import functions as F

从 pyspark.sql.类型导入数组类型， 双类型， 结构类型， 结构字段，
字符串类型， 长类型， 布尔类型

schema = StructType([
StructField("alertId", StringType(), True),
StructField("score", DoubleType(), True),
StructField("severity", DoubleType(), True),
])


from pyspark.sql.functions import from_json, to_json
from pyspark.sql.types import *
df = spark.readStream.format("eventhubs").options(**ehConf).load()
stream_data_body=
df.select(from_json(df.body.cast('string'),schema).\
alias('body')).select("body.*")

Answer 1

看起来您的有效负载中有一个记录列表，而不是单个记录。在这种情况下，您可以尝试以下模式：

schema = StructType([
  StructField("alerts", ArrayType(
    StructType([
      StructField("alertId", StringType(), True),
      StructField("score", DoubleType(), True),
      StructField("severity", DoubleType(), True),
    ])))])

然后为了得到单独的记录，你需要分解你的有效载荷——就像这样:

from pyspark.sql.functions import from_json, explode

stream_data = df.select(from_json(df.body.cast('string'),schema).alias('body')) \
  .select("body.*") \
  .select(explode("alerts").alias("alert")) \
  .select("alert.*")