假设 json_data 属于类型 map （您始终可以将其转换为 map 如果不是），你可以用 getItem :

df = spark.createDataFrame([
    [1, {"name": "abc", "depts": ["dep01", "dep02"]}],
    [2, {"name": "xyz", "depts": ["dep03"], "sal": 100}]
],
    ['id', 'json_data']
)

df.select(
    df.id, 
    df.json_data.getItem('name').alias('name'), 
    df.json_data.getItem('depts').alias('depts'), 
    df.json_data.getItem('sal').alias('sal')
).show()

+---+----+--------------+----+
| id|name|         depts| sal|
+---+----+--------------+----+
|  1| abc|[dep01, dep02]|null|
|  2| xyz|       [dep03]| 100|
+---+----+--------------+----+

更具动态性的列提取方法：

cols = ['name', 'depts', 'sal']
df.select(df.id, *(df.json_data.getItem(col).alias(col) for col in cols)).show()

基于@gaurang shah的回答，我实现了一个处理嵌套json结构的解决方案，并修复了使用单调递增的id（非顺序）的问题
在这种方法中，“populatecolumnname”函数递归地检查structtype列并填充列名。
“renamecolumns”函数通过将“.”替换为“\”来重命名列，以标识嵌套的json字段。
“addindex”函数在解析json列后向dataframe添加索引以加入dataframe。

def flattenJSON(df : DataFrame, columnName: String) : DataFrame = {

    val indexCol = "internal_temp_id"

    def populateColumnName(col : StructField) : Array[String] = {
        col.dataType match {
          case struct: StructType => struct.fields.flatMap(populateColumnName).map(col.name + "." + _)
          case rest         => Array(col.name)
        }
    }

    def renameColumns(name : String) : String = {
        if(name contains ".") {
            name + " as " + name.replaceAll("\\.", "_")
        }
        else name
    }

    def addIndex(df : DataFrame) : DataFrame = {

        // Append "rowid" column of type Long
        val newSchema = StructType(df.schema.fields ++ Array(StructField(indexCol, LongType, false)))

        // Zip on RDD level
        val rddWithId = df.rdd.zipWithIndex
        // Convert back to DataFrame
        spark.createDataFrame(rddWithId.map{ case (row, index) => Row.fromSeq(row.toSeq ++ Array(index))}, newSchema)
    }

    val dfWithID = addIndex(df)

    val jsonDF = df.select(columnName)

    val ds = jsonDF.rdd.map(_.getString(0)).toDS
    val parseDF = spark.read.option("inferSchema",true).json(ds)

    val columnNames = parseDF.schema.fields.flatMap(populateColumnName).map(renameColumns)

    var resultDF = parseDF.selectExpr(columnNames:_*)

    val jsonDFWithID = addIndex(resultDF)

    val joinDF = dfWithID.join(jsonDFWithID, indexCol).drop(indexCol)

    joinDF
}

val res = flattenJSON(jsonDF, "address")

如果它是一个 CSV 只有一列作为 JSON 数据。您可以使用以下解决方案。

val csvDF = spark.read.option("delimiter", "|").option("inferSchema", true).option("header", true).csv("test.csv")
val rdd = csvDF.select(" json_data").rdd.map(_.getString(0))
val ds = rdd.toDS
val jsonDF = spark.read.json(ds)
val jsonDFWithID = jsonDF.withColumn("id", monotonically_increasing_id())
val csvDFWithID = csvDF.select($"id ").withColumn("id", monotonically_increasing_id())
val joinDF = jsonDFWithID.join(csvDFWithID, "id").drop("id")

这就是最终Dataframe的样子。

scala> joinDF.printSchema()
root
 |-- address: struct (nullable = true)
 |    |-- city: string (nullable = true)
 |    |-- state: string (nullable = true)
 |-- depts: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- name: string (nullable = true)
 |-- sal: long (nullable = true)
 |-- id : double (nullable = true)

如果它是一个 JSON 文件。为了我。 inferSchema 很好用。
json文件

~/Downloads ▶ cat test.json
{"id": 1, "name":"abc", "depts":["dep01", "dep02"]},
{"id": 2, "name":"xyz", "depts" :["dep03"],"sal":100}

代码

scala> scc.read.format("json").option("inerSchema", true).load("Downloads/test.json").show()
+--------------+---+----+----+
|         depts| id|name| sal|
+--------------+---+----+----+
|[dep01, dep02]|  1| abc|null|
|       [dep03]|  2| xyz| 100|
+--------------+---+----+----+