我有两个由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 printschema():
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()
1条答案
按热度按时间zqry0prt1#
检查以下代码。
提取结构
Valori_Di_Borsa
列,添加News
列&重建结构。