Kemahiran meratakan untuk pelbagai struktur bersarang di Pyspark

Artikel ini meneroka secara mendalam bagaimana dengan cekap meratakan pelbagai lapisan bersarang kompleks (struct (array (struct)))) struktur ke dalam array (struct) dalam pyspark. Dengan menggabungkan Spark SQL mengubah fungsi-fungsi dan fungsi yang lebih tinggi, kami dapat mengekstrak medan struktur dalaman dan menggabungkannya dengan bidang luar, akhirnya mencapai pemudahan corak sasaran, mengelakkan kerumitan eksploit tradisional dan kombinasi kumpulan, dan menyediakan penyelesaian yang lebih baik dan berskala.

Memahami struktur dan matlamat bersarang yang kompleks

Apabila berurusan dengan data besar, kita sering menghadapi data data yang mengandungi jenis data bersarang kompleks. Senario biasa adalah lajur yang mengandungi struktur jenis jenis (struct (array (struct))))) seperti:

 akar
 |- A: Integer (nullable = benar)
 |- Senarai: Array (Nullable = Benar)
 | |- elemen: struct (containsNull = true)
 | | |- b: integer (nullable = true)
 | | |- sub_list: array (nullable = true)
 | | | |- elemen: struct (containsNull = true)
 | | | | |- c: integer (nullable = true)
 | | | | |- foo: string (nullable = true)

Matlamat kami adalah untuk mempermudahkan struktur bersarang berbilang lapisan ini ke dalam bentuk array (struct), iaitu, untuk mempromosikan bidang C dan Foo dalam sub_list ke struktur di dalam senarai, dan menghapuskan tahap bersarang sub_list:

 akar
 |- A: Integer (nullable = benar)
 |- Senarai: Array (Nullable = Benar)
 | |- elemen: struct (containsNull = true)
 | | |- b: integer (nullable = true)
 | | |- c: integer (nullable = true)
 | | |- Foo: String (Nullable True)

Proses meratakan ini adalah penting untuk analisis dan pemprosesan data berikutnya.

Cabaran dan batasan pendekatan tradisional

Kaedah meratakan tradisional biasanya melibatkan fungsi eksploit, yang memperluaskan setiap elemen dalam array ke dalam baris yang berasingan. Untuk struktur di atas, jika anda menggunakan mengeksploitasi secara langsung, ia mungkin mengambil pelbagai operasi mengeksploitasi dan kemudian menyusun semula melalui GroupBy dan Collect_list, yang boleh menjadi sangat kompleks dan tidak cekap dalam menghadapi sarang yang lebih dalam. Sebagai contoh, walaupun kaedah berikut adalah berkesan, mereka mahal untuk mengekalkan dalam senario kompleks:

 Dari Pyspark.SQL Import SparkSession
dari pyspark.sql.functions import dalam talian, expr, collect_list, struct

# Anggap DF adalah DataFrame anda
# df.Select ("a", inline ("senarai")) \
# .pelect (expr ("*"), inline ("sub_list")) \
# .drop ("sub_list") \
# .Groupby ("A") \
# .agg (collect_list (struct ("b", "c", "foo")). alias ("senarai"))

Pendekatan ini memerlukan kita untuk "mengangkat" semua tahap bersarang ke tahap baris dan kemudian agregat, yang bertentangan dengan falsafah transformasi "bottom-up" atau "di tempat" kami. Kami lebih suka penyelesaian yang boleh beroperasi secara langsung di dalam array tanpa mengubah bilangan baris data.

Penyelesaian Pyspark: Gabungan transformasi dan meratakan

Pyspark 3.x memperkenalkan fungsi pesanan yang lebih tinggi seperti Transform, yang sangat meningkatkan kuasa pemprosesan jenis data yang kompleks (terutamanya tatasusunan). Menggabungkan transformasi dan meratakan, kita dapat menyelesaikan masalah di atas dengan anggun.

Fungsi transformasi membolehkan kami menggunakan logik transformasi tersuai untuk setiap elemen dalam array dan mengembalikan array baru. Apabila ia datang kepada bersarang berbilang lapisan, kita boleh menggunakan transformasi bersarang untuk memproses lapisan mengikut lapisan.

Logik Teras :

1. Penukaran lapisan dalaman : Pertama, ubah sub_list paling dalam. Untuk setiap elemen dalam sub_list (iaitu, struktur yang mengandungi c dan foo), kami menggabungkannya dengan medan B di struktur luar untuk membuat struktur yang baru rata.
2. Penukaran Luar : Langkah transformasi ini akan menghasilkan struktur array (array (struct)).
3. FLATTEN : Akhirnya, gunakan fungsi meratakan untuk menggabungkan struktur array (array (struct)) ke dalam satu array (struct).

Contoh kod :

Pertama, kami membuat data simulasi untuk menunjukkan:

 Dari Pyspark.SQL Import SparkSession
dari pyspark.sql.functions import col, transform, flatten, struct
dari pyspark.sql.types Import StructType, Structfield, Arraytype, Integertype, StringType

# Inisialisasi SparkSession
Spark = SparkSession.Builder.AppName ("flattennestedArraystruct"). getOrcreate ()

# Tentukan skema awal
inner_struct_schema = structType ([
    StructField ("C", IntegerType (), benar),
    StructField ("foo", stringtype (), benar)
])
outer_struct_schema = structType ([
    StructField ("B", IntegerType (), benar),
    StructField ("sub_list", arraytype (inner_struct_schema), benar)
])
df_schema = structType ([
    StructField ("a", integertype (), benar),
    StructField ("Senarai", ArrayType (OUTER_STRUCT_SCHEMA), benar)
])

# Buat data sampel = [
    (1, [
        {"b": 10, "sub_list": [{"c": 100, "foo": "x"}, {"c": 101, "foo": "y"}}},
        {"b": 20, "sub_list": [{"c": 200, "foo": "z"}]}
    ]),
    (2, [[
        {"b": 30, "sub_list": [{"c": 300, "foo": "w"}]}
    ])
]

df = spark.createdataFrame (data, skema = df_schema)
df.printschema ()
df.show (truncate = false)

# Memohon logik yang diratakan df_flattened = df.withColumn (
    "Senarai",
    meratakan (
        transformasi (
            col ("senarai"), # array luar (pelbagai struktur)
            Lambda X: Transform ( # beroperasi pada setiap struct x array luar x.getfield ("sub_list"), # get sub_list (array structs) dalam struct x
                lambda y: struct (x.getfield ("b"). alias ("b"), y.getfield ("c") alias ("c"), y.getfield ("foo") alias ("foo")),
            ))
        )
    ))
)

df_flattened.printschema ()
df_flattened.show (truncate = false)

# Hentikan SparkSession
Spark.stop ()

Kod parsing

1. df.WithColumn ("Senarai", ...): Kami memilih untuk mengubah suai lajur senarai untuk mengandungi hasil yang diratakan.
2. Transform (Col ("Senarai"), Lambda X: ...): Ini adalah transformasi luar. Ia melangkah ke atas setiap elemen Struct dalam lajur Senarai, yang kami namakan sebagai x. Jenis x adalah struct (b: int, sub_list: array (struct (c: int, foo: string))).
3. Transform (X.GetField ("Sub_list"), Lambda Y: ...): Ini adalah transformasi dalaman. Ia bertindak di medan sub_list dalam x. sub_list adalah array yang unsur -unsur (struktur (c: int, foo: string)) dinamakan y.
4. struct (x.getfield ("b") alias ("b"), y.getfield ("c") alias ("c"), y.getfield ("foo") alias ("foo")): Di dalam transformasi dalaman, kami membina struktur baru.
   • X.GetField ("B"): Dapatkan medan B dari Struct Out X.
   • Y.Getfield ("C"): Dapatkan medan C dari struktur dalaman y.
   • y.getfield ("foo"): Dapatkan medan foo dari struktur dalaman y.
   • alias ("b"), alias ("c"), alias ("foo") digunakan untuk memastikan bahawa nama medan struct yang baru dijana adalah betul. Fungsi struct ini menghasilkan struktur yang diratakan untuk setiap elemen y dalam sub_list. Oleh itu, hasil transformasi dalaman adalah array (struct (b: int, c: int, foo: string)).
5. Hasil pertengahan : Transformasi luar mengumpul semua tatasusunan ini (struct), jadi output terakhirnya adalah array (array (struct (b: int, c: int, foo: string))).
6. Flatten (...): Akhirnya, fungsi rata -rata meratakan struktur array (array (struct)) ke dalam satu array (struct (b: int, c: int, foo: string)), yang betul -betul apa yang kita harapkan untuk menargetkan skema.

Nota dan amalan terbaik

• Nama medan : Pastikan nama medan yang digunakan dalam Getfield () dan struct () betul -betul sepadan dengan nama dalam skema sebenar.
• Pemprosesan nilai null : Fungsi transformasi secara semulajadi memproses elemen kosong dalam array. Jika sub_list kosong, transformasi dalaman mengembalikan array kosong; Jika senarai kosong, transformasi luar juga mengembalikan array kosong. Flatten mengendalikan tatasusunan kosong juga selamat.
• Prestasi : Transform adalah fungsi pesanan tinggi terbina dalam untuk Spark SQL, yang biasanya mempunyai prestasi yang lebih baik daripada UDFS adat (fungsi yang ditentukan oleh pengguna) kerana ia dapat dioptimumkan dalam Spark Catalyst Optimizer.
• Kebolehbacaan : Walaupun transformasi bersarang sangat kuat, over-necking dapat mengurangkan kebolehbacaan kod. Untuk senario yang lebih kompleks, pertimbangkan untuk memisahkan logik transformasi ke dalam pelbagai langkah atau menambah komen terperinci.
• Universalitas : Corak rata yang digabungkan dengan transformasi ini dapat disebarkan kepada struktur bersarang yang lebih dalam, hanya meningkatkan tahap transformasi bersarang.

Meringkaskan

Dengan bijak menggabungkan fungsi transformasi dan meratakan Pyspark, kami dapat meratakan pelbagai struktur bersarang pelbagai (struct (array (struct))) ke dalam struktur array (struct) yang lebih mudah untuk proses dalam cara deklaratif dan cekap. Pendekatan ini menghindari kerumitan eksploit tradisional dan kombinasi kumpulan, dan sangat sesuai untuk senario di mana unsur -unsur yang berubah -ubah di dalam tatasusunan diperlukan. Ia adalah helah yang sangat berguna apabila berurusan dengan data separuh berstruktur yang kompleks dalam Spark.

Atas ialah kandungan terperinci Kemahiran meratakan untuk pelbagai struktur bersarang di Pyspark. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!