python - स्पार्क डेटाफ़्रेम में नेस्टेड फ़ील्ड का नाम बदलें

apache-spark dataframe (1)

स्पार्क में डेटाफ्रेम df होने:

 |-- array_field: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: string (nullable = true)
 |    |    |-- b: long (nullable = true)
 |    |    |-- c: long (nullable = true)

फ़ील्ड array_field.a को array_field.a_renamed कैसे बदला array_field.a_renamed ?

[अद्यतन करें]:

.withColumnRenamed() नेस्टेड फ़ील्ड्स के साथ काम नहीं करता है, इसलिए मैंने इस हैक किए गए और असुरक्षित तरीके की कोशिश की:

# First alter the schema:
schema = df.schema
schema['array_field'].dataType.elementType['a'].name = 'a_renamed'

ind = schema['array_field'].dataType.elementType.names.index('a')
schema['array_field'].dataType.elementType.names[ind] = 'a_renamed'

# Then set dataframe's schema with altered schema
df._schema = schema

मुझे पता है कि एक निजी विशेषता सेट करना एक अच्छा अभ्यास नहीं है, लेकिन मैं df के लिए स्कीमा सेट करने का कोई अन्य तरीका नहीं जानता

मुझे लगता है कि मैं एक सही रास्ते पर df.printSchema() लेकिन df.printSchema() अभी भी array_field.a लिए पुराना नाम array_field.a , हालांकि df.schema == schema True

अजगर

एकल नेस्टेड फ़ील्ड को संशोधित करना संभव नहीं है। आपको एक पूरी संरचना को फिर से बनाना होगा। इस विशेष मामले में सबसे आसान समाधान cast का उपयोग करना है।

पहले आयात का एक गुच्छा:

from collections import namedtuple
from pyspark.sql.functions import col
from pyspark.sql.types import (
    ArrayType, LongType, StringType, StructField, StructType)

और उदाहरण डेटा:

Record = namedtuple("Record", ["a", "b", "c"])

df = sc.parallelize([([Record("foo", 1, 3)], )]).toDF(["array_field"])

आइए पुष्टि करते हैं कि स्कीमा आपके मामले में समान है:

df.printSchema()

root
 |-- array_field: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: string (nullable = true)
 |    |    |-- b: long (nullable = true)
 |    |    |-- c: long (nullable = true)

आप एक स्ट्रिंग के रूप में उदाहरण के लिए एक नया स्कीमा परिभाषित कर सकते हैं:

str_schema = "array<struct<a_renamed:string,b:bigint,c:bigint>>"

df.select(col("array_field").cast(str_schema)).printSchema()

root
 |-- array_field: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a_renamed: string (nullable = true)
 |    |    |-- b: long (nullable = true)
 |    |    |-- c: long (nullable = true)

या एक DataType :

struct_schema = ArrayType(StructType([
    StructField("a_renamed", StringType()),
    StructField("b", LongType()),
    StructField("c", LongType())
]))

 df.select(col("array_field").cast(struct_schema)).printSchema()

root
 |-- array_field: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a_renamed: string (nullable = true)
 |    |    |-- b: long (nullable = true)
 |    |    |-- c: long (nullable = true)

स्काला

स्काला में समान तकनीकों का उपयोग किया जा सकता है:

case class Record(a: String, b: Long, c: Long)

val df = Seq(Tuple1(Seq(Record("foo", 1, 3)))).toDF("array_field")

val strSchema = "array<struct<a_renamed:string,b:bigint,c:bigint>>"

df.select($"array_field".cast(strSchema))

या

import org.apache.spark.sql.types._

val structSchema = ArrayType(StructType(Seq(
    StructField("a_renamed", StringType),
    StructField("b", LongType),
    StructField("c", LongType)
)))

df.select($"array_field".cast(structSchema))

संभावित सुधार :

यदि आप एक अभिव्यंजक डेटा हेरफेर या JSON प्रोसेसिंग लाइब्रेरी का उपयोग करते हैं, तो डेटा प्रकारों को dict या JSON स्ट्रिंग में डंप करना और उदाहरण के लिए इसे वहां से ले जाना आसान हो सकता है (पायथन / toolz ):

from toolz.curried import pipe, assoc_in, update_in, map
from operator import attrgetter

# Update name to "a_updated" if name is "a"
rename_field = update_in(
    keys=["name"], func=lambda x: "a_updated" if x == "a" else x)

updated_schema = pipe(
   #  Get schema of the field as a dict
   df.schema["array_field"].jsonValue(),
   # Update fields with rename
   update_in(
       keys=["type", "elementType", "fields"],
       func=lambda x: pipe(x, map(rename_field), list)),
   # Load schema from dict
   StructField.fromJson,
   # Get data type
   attrgetter("dataType"))

df.select(col("array_field").cast(updated_schema)).printSchema()