时间戳

首页 > 解决方案 > 如何更改 pyspark 上的 JSON 结构？

python - 如何更改 pyspark 上的 JSON 结构？

问题描述

我有两个由 kafka 读取的 json 文件，这是他们的 printSchema ()

JSON1 打印模式：

root
 |-- _id: string (nullable = true)
 |-- Data: string (nullable = true)
 |-- NomeAzienda: string (nullable = true)
 |-- Valori_Di_Borsa: struct (nullable = false)
 |    |-- PrezzoUltimoContratto: double (nullable = true)
 |    |-- Var%: double (nullable = true)
 |    |-- VarAssoluta: double (nullable = true)
 |    |-- OraUltimoContratto: string (nullable = true)
 |    |-- QuantitaUltimo: double (nullable = true)
 |    |-- QuantitaAcquisto: double (nullable = true)
 |    |-- QuantitaVendita: double (nullable = true)
 |    |-- QuantitaTotale: double (nullable = true)
 |    |-- NumeroContratti: double (nullable = true)
 |    |-- MaxOggi: double (nullable = true)
 |    |-- MinOggi: double (nullable = true)

JSON2 打印模式（）：

root
 |-- _id: string (nullable = true)
 |-- News: struct (nullable = false)
 |    |-- TitoloNews: string (nullable = true)
 |    |-- TestoNews: string (nullable = true)
 |    |-- DataNews: string (nullable = true)
 |    |-- OraNews: long (nullable = true)
 |    |-- SoggettoNews: string (nullable = true)

加入两个 JSON，我得到了这个 printSchema()：

root
 |-- _id: string (nullable = true)
 |-- Data: string (nullable = true)
 |-- NomeAzienda: string (nullable = true)
 |-- Valori_Di_Borsa: struct (nullable = false)
 |    |-- PrezzoUltimoContratto: double (nullable = true)
 |    |-- Var%: double (nullable = true)
 |    |-- VarAssoluta: double (nullable = true)
 |    |-- OraUltimoContratto: string (nullable = true)
 |    |-- QuantitaUltimo: double (nullable = true)
 |    |-- QuantitaAcquisto: double (nullable = true)
 |    |-- QuantitaVendita: double (nullable = true)
 |    |-- QuantitaTotale: double (nullable = true)
 |    |-- NumeroContratti: double (nullable = true)
 |    |-- MaxOggi: double (nullable = true)
 |    |-- MinOggi: double (nullable = true)
 |-- _id: string (nullable = true)
 |-- News: struct (nullable = false)
 |    |-- TitoloNews: string (nullable = true)
 |    |-- TestoNews: string (nullable = true)
 |    |-- DataNews: string (nullable = true)
 |    |-- OraNews: long (nullable = true)
 |    |-- SoggettoNews: string (nullable = true)

但我想要的结果是这样的：

更新根：

 -- _id: string (nullable = true)
 -- Data: string (nullable = true)
 -- NomeAzienda: string (nullable = true)
 -- Valori_Di_Borsa: struct (nullable = false)
     |-- PrezzoUltimoContratto: double (nullable = true)
     |-- Var%: double (nullable = true)
     |-- VarAssoluta: double (nullable = true)
     |-- OraUltimoContratto: string (nullable = true)
     |-- QuantitaUltimo: double (nullable = true)
     |-- QuantitaAcquisto: double (nullable = true)
     |-- QuantitaVendita: double (nullable = true)
     |-- QuantitaTotale: double (nullable = true)
     |-- NumeroContratti: double (nullable = true)
     |-- MaxOggi: double (nullable = true)
     |-- MinOggi: double (nullable = true)
     |-- News: struct (nullable = false)
                |-- id: string (nullable = true)
                |-- TitoloNews: string (nullable = true)
                |-- TestoNews: string (nullable = true)
                |-- DataNews: string (nullable = true)
                |-- OraNews: long (nullable = true)
                |-- SoggettoNews: string (nullable = true)

我怎样才能使用 pyspark 做到这一点？

这是我的代码：

   df_borsa = spark.readStream.format("kafka") \
                  .option("kafka.bootstrap.servers", kafka_broker) \
                  .option("startingOffsets", "latest") \
                  .option("subscribe","Be_borsa") \
                  .load() \
                  .selectExpr("CAST(value AS STRING)") 

   df_news = spark.readStream.format("kafka") \
                  .option("kafka.bootstrap.servers", kafka_broker) \
                  .option("startingOffsets", "latest") \
                  .option("subscribe","Ita_news") \
                  .load() \
                  .selectExpr("CAST(value AS STRING)") 

    df_borsa =df_borsa.withColumn("Valori_Di_Borsa",F.struct(F.col("PrezzoUltimoContratto"),F.col("Var%"),F.col("VarAssoluta"),F.col("OraUltimoContratto"),F.col("QuantitaUltimo"),F.col("QuantitaAcquisto"),F.col("QuantitaVendita"),F.col("QuantitaTotale"),F.col("NumeroContratti"),F.col("MaxOggi"),F.col("MinOggi")))

    df_borsa.printSchema()

    df_news = df_news.withColumn("News",F.struct(F.col("TitoloNews"),F.col("TestoNews"),F.col("DataNews"),F.col("OraNews"),F.col("SoggettoNews")))

    df_news.printSchema()

    df_join = df_borsa.join(df_news)

    df_join.printSchema()

标签： pythonjsonapache-sparkpysparkapache-kafka

解决方案

检查下面的代码。

提取结构Valori_Di_Borsa列，添加News列并重建结构。

df_join = df_borsa.join(df_news)
.withColumn("Valori_Di_Borsa",F.struct(F.col("Valori_Di_Borsa.*"),F.col("News"))))

推荐阅读