将pyspark中嵌套的Dataframe展平到列中

n3h0vuf2  于 2021-05-24  发布在  Spark
关注(0)|答案(2)|浏览(580)

嗨，我有json数据，我拉在pyspark的样本如下。

{
    "data": [
        ["row-r9pv-p86t.ifsp", "00000000-0000-0000-0838-60C2FFCC43AE", 0, 1574264158, null, 1574264158, null, "{ }", "2007", "ZOEY", "KINGS", "F", "11"],
        ["row-7v2v~88z5-44se", "00000000-0000-0000-C8FC-DDD3F9A72DFF", 0, 1574264158, null, 1574264158, null, "{ }", "2007", "ZOEY", "SUFFOLK", "F", "6"],
        ["row-hzc9-4kvv~mbc9", "00000000-0000-0000-562E-D9A0792557FC", 0, 1574264158, null, 1574264158, null, "{ }", "2007", "ZOEY", "MONROE", "F", "6"]
    ]
}

我试图分解多数组，并将每条记录分解为一行Dataframe，但看起来是这样的：

df= spark.read.json('data/rows.json', multiLine=True)
temp_df = df.select(explode("data").alias("data"))
temp_df.show(n=3, truncate=False)

结果：

+-----------------------------------------------------------------------------------------------------------------------+
|data                                                                                                                   |
+-----------------------------------------------------------------------------------------------------------------------+
|[row-r9pv-p86t.ifsp, 00000000-0000-0000-0838-60C2FFCC43AE, 0, 1574264158,, 1574264158,, { }, 2007, ZOEY, KINGS, F, 11] |
|[row-7v2v~88z5-44se, 00000000-0000-0000-C8FC-DDD3F9A72DFF, 0, 1574264158,, 1574264158,, { }, 2007, ZOEY, SUFFOLK, F, 6]|
|[row-hzc9-4kvv~mbc9, 00000000-0000-0000-562E-D9A0792557FC, 0, 1574264158,, 1574264158,, { }, 2007, ZOEY, MONROE, F, 6] |
+-----------------------------------------------------------------------------------------------------------------------+
temp_df.printSchema()
temp_df.show(5)
temp_df.select(flatten(temp_df.data)).show(n=10)

到目前为止还不错，但是当我尝试使用 flatten 方法它给了我错误的说法 cannot resolve 'flatten('data')' due to data type mismatch: The argument should be an array of arrays, but 'data' is of array<string> type. 这是有道理的，但我不知道我们如何才能平展阵列。
我应该写一些习惯吗 map 方法将行数组Map到Dataframe列？

DataFrameapache-sparkpysparkapache-spark-sql

2条答案

按热度按时间
fsi0uk1n

fsi0uk1n1#

val resDF = temp_df.select(
  'data.getItem(0).alias("c0"),
  'data.getItem(1).alias("c1"),
  'data.getItem(2).alias("c2"),
  'data.getItem(3).alias("c3")
  // ...
)
resDF.show(false)
//  +------------------+------------------------------------+---+----------+
//  |c0                |c1                                  |c2 |c3        |
//  +------------------+------------------------------------+---+----------+
//  |row-r9pv-p86t.ifsp|00000000-0000-0000-0838-60C2FFCC43AE|0  |1574264158|
//  |row-7v2v~88z5-44se|00000000-0000-0000-C8FC-DDD3F9A72DFF|0  |1574264158|
//  |row-hzc9-4kvv~mbc9|00000000-0000-0000-562E-D9A0792557FC|0  |1574264158|
//  +------------------+------------------------------------+---+----------+

v 2（与Column和concat一起使用）：

val sourceDF = Seq(
    Array("row-r9pv-p86t.ifsp", "00000000-0000-0000-0838-60C2FFCC43AE", "0", "1574264158", "", "1574264158", "", "{ }", "2007", "ZOEY", "KINGS", "F", "11"),
    Array("row-7v2v~88z5-44se", "00000000-0000-0000-C8FC-DDD3F9A72DFF", "0", "1574264158", "", "1574264158", "", "{ }", "2007", "ZOEY", "SUFFOLK", "F", "6"),
    Array("row-hzc9-4kvv~mbc9", "00000000-0000-0000-562E-D9A0792557FC", "0", "1574264158", "", "1574264158", "", "{ }", "2007", "ZOEY", "MONROE", "F", "6")
  ).toDF("dataColumn")

  sourceDF.show(false)

//  +-------------------------------------------------------------------------------------------------------------------------+
//  |dataColumn                                                                                                               |
//  +-------------------------------------------------------------------------------------------------------------------------+
//  |[row-r9pv-p86t.ifsp, 00000000-0000-0000-0838-60C2FFCC43AE, 0, 1574264158, , 1574264158, , { }, 2007, ZOEY, KINGS, F, 11] |
//  |[row-7v2v~88z5-44se, 00000000-0000-0000-C8FC-DDD3F9A72DFF, 0, 1574264158, , 1574264158, , { }, 2007, ZOEY, SUFFOLK, F, 6]|
//  |[row-hzc9-4kvv~mbc9, 00000000-0000-0000-562E-D9A0792557FC, 0, 1574264158, , 1574264158, , { }, 2007, ZOEY, MONROE, F, 6] |
//  +-------------------------------------------------------------------------------------------------------------------------+

  val df1 = sourceDF
    .withColumn("dataString", concat_ws(", ", 'dataColumn))
    .select('dataString)

  df1.printSchema()

  df1.show(false)
//  root
//  |-- dataString: string (nullable = false)
//
//  +-----------------------------------------------------------------------------------------------------------------------+
//  |dataString                                                                                                             |
//  +-----------------------------------------------------------------------------------------------------------------------+
//  |row-r9pv-p86t.ifsp, 00000000-0000-0000-0838-60C2FFCC43AE, 0, 1574264158, , 1574264158, , { }, 2007, ZOEY, KINGS, F, 11 |
//  |row-7v2v~88z5-44se, 00000000-0000-0000-C8FC-DDD3F9A72DFF, 0, 1574264158, , 1574264158, , { }, 2007, ZOEY, SUFFOLK, F, 6|
//  |row-hzc9-4kvv~mbc9, 00000000-0000-0000-562E-D9A0792557FC, 0, 1574264158, , 1574264158, , { }, 2007, ZOEY, MONROE, F, 6 |
//  +-----------------------------------------------------------------------------------------------------------------------+

  val df2 = df1.select(
    split('dataString, ", ").getItem(0).alias("c0"),
    split('dataString, ", ").getItem(1).alias("c1"),
    split('dataString, ", ").getItem(2).alias("c2"),
    split('dataString, ", ").getItem(3).alias("c3"),
    split('dataString, ", ").getItem(4).alias("c4"),
    split('dataString, ", ").getItem(5).alias("c5"),
    split('dataString, ", ").getItem(6).alias("c6"),
    split('dataString, ", ").getItem(7).alias("c7"),
    split('dataString, ", ").getItem(8).alias("c8"),
    split('dataString, ", ").getItem(9).alias("c9"),
    split('dataString, ", ").getItem(10).alias("c10"),
    split('dataString, ", ").getItem(11).alias("c11"),
    split('dataString, ", ").getItem(12).alias("c12")
  )
  df2.printSchema()
//  root
//  |-- c0: string (nullable = true)
//  |-- c1: string (nullable = true)
//  |-- c2: string (nullable = true)
//  |-- c3: string (nullable = true)
//  |-- c4: string (nullable = true)
//  |-- c5: string (nullable = true)
//  |-- c6: string (nullable = true)
//  |-- c7: string (nullable = true)
//  |-- c8: string (nullable = true)
//  |-- c9: string (nullable = true)
//  |-- c10: string (nullable = true)
//  |-- c11: string (nullable = true)
//  |-- c12: string (nullable = true)

  df2.show(false)
//  +------------------+------------------------------------+---+----------+---+----------+---+---+----+----+-------+---+---+
//  |c0                |c1                                  |c2 |c3        |c4 |c5        |c6 |c7 |c8  |c9  |c10    |c11|c12|
//  +------------------+------------------------------------+---+----------+---+----------+---+---+----+----+-------+---+---+
//  |row-r9pv-p86t.ifsp|00000000-0000-0000-0838-60C2FFCC43AE|0  |1574264158|   |1574264158|   |{ }|2007|ZOEY|KINGS  |F  |11 |
//  |row-7v2v~88z5-44se|00000000-0000-0000-C8FC-DDD3F9A72DFF|0  |1574264158|   |1574264158|   |{ }|2007|ZOEY|SUFFOLK|F  |6  |
//  |row-hzc9-4kvv~mbc9|00000000-0000-0000-562E-D9A0792557FC|0  |1574264158|   |1574264158|   |{ }|2007|ZOEY|MONROE |F  |6  |
//  +------------------+------------------------------------+---+----------+---+----------+---+---+----+----+-------+---+---+
赞(0）分享  回复(0）举报  2021-05-25
vcirk6k6

vcirk6k62#

回答我自己的问题。所以它可以帮助任何需要帮助的人。
从文件中读取源数据

df= spark.read.json('data/rows.json', multiLine=True)
temp_df = df.select(explode("data").alias("data"))
temp_df.show(n=3, truncate=False)

结果：

+-----------------------------------------------------------------------------------------------------------------------+
|data                                                                                                                   |
+-----------------------------------------------------------------------------------------------------------------------+
|[row-r9pv-p86t.ifsp, 00000000-0000-0000-0838-60C2FFCC43AE, 0, 1574264158,, 1574264158,, { }, 2007, ZOEY, KINGS, F, 11] |
|[row-7v2v~88z5-44se, 00000000-0000-0000-C8FC-DDD3F9A72DFF, 0, 1574264158,, 1574264158,, { }, 2007, ZOEY, SUFFOLK, F, 6]|
|[row-hzc9-4kvv~mbc9, 00000000-0000-0000-562E-D9A0792557FC, 0, 1574264158,, 1574264158,, { }, 2007, ZOEY, MONROE, F, 6] |
+-----------------------------------------------------------------------------------------------------------------------+

在上面的数据框中，每个单元格都包含一个字符串数组，我需要的是单独列中的每个元素和特定的数据类型。

df = temp_df.withColumn("sid", temp_df["data"].getItem(0).cast(StringType())) \
       .withColumn("id", temp_df["data"].getItem(1).cast(IntegerType())) \
       .withColumn("position", temp_df["data"].getItem(2).cast(IntegerType())) \
       .withColumn("created_at", temp_df["data"].getItem(3).cast(TimestampType())) \
       .withColumn("created_meta", temp_df["data"].getItem(4).cast(StringType())) \
       .withColumn("updated_at", temp_df["data"].getItem(5).cast(TimestampType())) \
       .withColumn("updated_meta", temp_df["data"].getItem(6).cast(StringType())) \
       .withColumn("meta", temp_df["data"].getItem(7).cast(StringType())) \
       .withColumn("Year", (temp_df["data"].getItem(8)).cast(IntegerType())) \
       .withColumn("First Name", temp_df["data"].getItem(9).cast(StringType())) \
       .withColumn("County", temp_df["data"].getItem(10).cast(StringType())) \
       .withColumn("Sex", temp_df["data"].getItem(11).cast(StringType())) \
       .withColumn("Count", temp_df["data"].getItem(12).cast(IntegerType())) \
       .drop("data")
df.show()
df.printSchema()
+------------------+----+--------+----------+------------+----------+------------+----+----+----------+-------+---+-----+
|               sid|  id|position|created_at|created_meta|updated_at|updated_meta|meta|Year|First Name| County|Sex|Count|
+------------------+----+--------+----------+------------+----------+------------+----+----+----------+-------+---+-----+
|row-r9pv-p86t.ifsp|null|       0|      null|        null|      null|        null| { }|2007|      ZOEY|  KINGS|  F|   11|
|row-7v2v~88z5-44se|null|       0|      null|        null|      null|        null| { }|2007|      ZOEY|SUFFOLK|  F|    6|
|row-hzc9-4kvv~mbc9|null|       0|      null|        null|      null|        null| { }|2007|      ZOEY| MONROE|  F|    6|
+------------------+----+--------+----------+------------+----------+------------+----+----+----------+-------+---+-----+

==================== SCHEMA ====================

root
 |-- sid: string (nullable = true)
 |-- id: integer (nullable = true)
 |-- position: integer (nullable = true)
 |-- created_at: timestamp (nullable = true)
 |-- created_meta: string (nullable = true)
 |-- updated_at: timestamp (nullable = true)
 |-- updated_meta: string (nullable = true)
 |-- meta: string (nullable = true)
 |-- Year: integer (nullable = true)
 |-- First Name: string (nullable = true)
 |-- County: string (nullable = true)
 |-- Sex: string (nullable = true)
 |-- Count: integer (nullable = true)
赞(0）分享  回复(0）举报  2021-05-25
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com