Pyspark의 다층 중첩 어레이 구조를위한 평평한 기술

이 기사는 Pyspark의 복잡한 다층 중첩 어레이 (struct (array (struct)) 구조를 배열 (struct)로 효율적으로 평평하게 평평하게 평평하게 평평하게하는 방법을 자세히 살펴 봅니다. Spark SQL의 변환 고차 함수와 평평한 기능을 결합함으로써 내부 구조 필드를 우아하게 추출하여 외부 필드와 병합하여 궁극적으로 목표 패턴의 단순화를 달성하고 전통적인 익스플로잇 및 그룹 조합의 복잡성을 피하고보다 선언적이고 확장 가능한 솔루션을 제공 할 수 있습니다.

복잡한 중첩 구조와 목표를 이해하십시오

빅 데이터를 다룰 때 복잡한 중첩 데이터 유형을 포함하는 데이터 프레임이 종종 발생합니다. 일반적인 시나리오는 유형 배열 구조 (struct (array (struct)))의 구조를 포함하는 열입니다.

 뿌리
 |- a : 정수 (nullable = true)
 |- 목록 : 배열 (nullable = true)
 | |- 요소 : struct (containsNull = true)
 | | |- b : 정수 (nullable = true)
 | | |- sub_list : 배열 (nullable = true)
 | | | |- 요소 : struct (containsNull = true)
 | | | | |- C : 정수 (nullable = true)
 | | | | |- foo : String (nullable = true)

우리의 목표는이 멀티 레이어 중첩 구조를 배열 (struct) 형태로 단순화하는 것입니다.

 뿌리
 |- a : 정수 (nullable = true)
 |- 목록 : 배열 (nullable = true)
 | |- 요소 : struct (containsNull = true)
 | | |- b : 정수 (nullable = true)
 | | |- C : 정수 (nullable = true)
 | | |- foo : String (nullable true)

이 평평한 프로세스는 후속 데이터 분석 및 처리에 중요합니다.

전통적인 접근 방식의 도전과 한계

기존의 평평한 방법에는 일반적으로 익스플로잇 기능이 포함되며,이 기능은 배열의 각 요소를 별도의 행으로 확장합니다. 위의 구조의 경우 Exploit을 직접 사용하는 경우 여러 개의 익스플로잇 작업이 필요하고 Groupby 및 Collect_list를 통해 Reggreggate가 발생하여 더 깊은 둥지에 직면하여 매우 복잡하고 비효율적 일 수 있습니다. 예를 들어, 다음 방법은 효과적이지만 복잡한 시나리오에서 유지하는 데 비용이 많이 듭니다.

 Pyspark.SQL 가져 오기 Sparksession에서
pyspark.sql.functions import inline, expr, collect_list, struct

# DF가 귀하의 데이터 프레임이라고 가정합니다
# df.select ( "a", inline ( "list")) \
# .select (expr ( "*"), inline ( "sub_list") \
# .Drop ( "sub_list") \
# .Groupby ( "A") \
# .Agg (collect_list (struct ( "b", "c", "foo")). 별명 ( "목록"))

이 접근법은 우리가 모든 중첩 레벨을 행 수준으로 "높이"다음 집계를 "상승"해야하며, 이는 예상되는 "상향식"또는 "내 현장"변환 철학에 위배됩니다. 데이터 프레임 행 수를 변경하지 않고 배열 내부에서 직접 작동 할 수있는 솔루션을 선호합니다.

Pyspark 솔루션 : 변환과 평평한 조합

PySpark 3.X는 변환과 같은 고차 기능을 도입하여 복잡한 데이터 유형 (특히 어레이)의 처리 능력을 크게 향상시킵니다. 변형과 평평한 결합으로 위의 문제를 우아하게 해결할 수 있습니다.

변환 함수를 사용하면 배열의 각 요소에 사용자 정의 변환 로직을 적용하고 새 배열을 반환 할 수 있습니다. 다층 둥지와 관련하여 중첩 변환을 사용하여 레이어별로 레이어를 처리 할 수 ​​있습니다.

핵심 논리 :

1. 내부 레이어 변환 : 먼저 가장 안쪽의 sub_list를 변환하십시오. sub_list의 각 요소에 대해 (즉, C 및 FOO를 포함하는 구조물)에 대해 외부 구조물의 B 필드와 결합하여 새로운 평평한 구조물을 만듭니다.
2. 외부 변환 :이 변환 단계는 배열 (Array (struct)) 구조를 생성합니다.
3. 평평한 : 마지막으로 평평한 기능을 사용하여 배열 (Array (struct)) 구조를 단일 어레이 (struct)로 병합하십시오.

샘플 코드 :

먼저, 우리는 시뮬레이션 된 데이터 프레임을 만듭니다.

 Pyspark.SQL 가져 오기 Sparksession에서
pyspark.sql.functions import col, transform, 평평한, struct
pyspark.sql.types import structtype, structfield, arraytype, integertype, stringtype

# Sparksession 초기화
spark = sparksession.builder.appname ( "flattennestedArraystruct"). getorcreate ()

# 초기 스키마를 정의합니다
inner_struct_schema = structtype ([[
    structfield ( "c", integertype (), true),
    structfield ( "foo", stringtype (), true)
])))
outer_struct_schema = structtype ([[
    structfield ( "b", integertype (), true),
    structfield ( "sub_list", arraytype (inner_struct_schema), true)
])))
df_schema = structType ([
    structfield ( "a", integertype (), true),
    structfield ( "목록", Arraytype (outer_struct_schema), true)
])))

# 샘플 데이터 생성 = [
    (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, schema = df_schema)
df.printschema ()
df.show (truncate = false)

# 평평한 로직 DF_FLATTENED = DF.WITHCOLUMN (
    "목록",
    단조롭게 하다(
        변환(
            col ( "list"), # 외부 배열 (배열)
            lambda x : transform ( # 외부 배열의 각 구조 X에서 작동합니다 x.getfield ( "sub_list"), # struct x의 get sub_list (배열)
                lambda y : struct (x.getfield ( "b"). 별칭 ( "b"), y.getfield ( "c"). alias ( "c"), y.getfield ( "foo"). alias ( "foo")),
            ),)
        ))
    ),)
))

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

# 스파크를 중지합니다
spark.stop ()

코드 구문 분석

1. df.withcolumn ( "list", ...) : 평평한 결과를 포함하도록 목록 열을 수정하기로 선택합니다.
2. transform (col ( "list"), lambda x : ...) : 이것은 외부 변환입니다. 목록 열의 각 구조 요소 위에 반복되며 x라고합니다. X 유형은 struct (b : int, sub_list : array (struct (c : int, foo : string))입니다.
3. transform (x.getfield ( "sub_list"), lambda y : ...) : 이것은 내부 변환입니다. x의 sub_list 필드에서 작용합니다. sub_list는 요소 (struct (c : int, foo : string))의 이름이 y라는 배열입니다.
4. struct (x.getfield ( "b"). alias ( "b"), y.getfield ( "c"). alias ( "c"), y.getfield ( "foo"). alias ( "foo")) : 내부 변환 내부에 새로운 구조물을 구축합니다.
   • X.GetField ( "B") : OUTER STRUCT X에서 B 필드를 가져옵니다.
   • y.getfield ( "C") : 내부 구조물 y에서 C 필드를 가져옵니다.
   • y.getfield ( "foo") : 내부 구조물 y에서 foo 필드를 가져옵니다.
   • 별명 ( "B"), 별칭 ( "C"), 별명 ( "FOO")이 새로 생성 된 구조장 필드 이름이 올바른지 확인하는 데 사용됩니다. 이 구조물 함수는 sub_list의 각 y 요소에 대해 평평한 구조를 생성합니다. 따라서 내부 변환의 결과는 배열입니다 (struct (b : int, c : int, foo : string)).
5. 중간 결과 : 외부 변환은 이러한 모든 배열 (struct)을 수집하므로 최종 출력은 배열입니다 (구조 (b : int, c : int, foo : foo : string)).
6. 평평한 (...) : 마지막으로, 평평한 함수는 배열 (Array (struct)) 구조를 단일 배열 (struct (b : int, c : int, foo : string))로 평평하게합니다.

메모 및 모범 사례

• 필드 이름 : getfield () 및 struct ()에 사용 된 필드 이름이 실제 스키마의 이름과 정확히 일치하는지 확인하십시오.
• 널 값 처리 : 변환 함수는 배열에서 빈 요소를 자연스럽게 처리합니다. sub_list가 비어 있으면 내부 변환은 빈 배열을 반환합니다. 목록이 비어 있으면 외부 변환은 빈 배열도 반환합니다. 평평한 손잡이 빈 배열도 안전합니다.
• 성능 : Transform은 Spark Catalyst Optimizer에서 최적화 할 수 있기 때문에 일반적으로 사용자 정의 UDF (사용자 정의 함수)보다 성능이 향상되는 Spark SQL의 내장 고차 기능입니다.
• 가독성 : 중첩 변환은 매우 강력하지만 막힘은 코드의 가독성을 줄일 수 있습니다. 보다 복잡한 시나리오의 경우 변환 로직을 여러 단계로 나누거나 자세한 주석을 추가하는 것을 고려하십시오.
• 유니버설 :이 변형-흡착 된 평평한 패턴은 더 깊은 둥지 구조로 일반화 될 수 있으며 중첩 수준을 증가시킬 수 있습니다.

요약

Pyspark의 변환 및 평평한 기능을 영리하게 결합함으로써 복잡한 다층 중첩 어레이 (struct (array)) 구조를 선언적이고 효율적인 방식으로 프로세스하기 쉬운 배열 (구조) 구조로 평평하게 할 수 있습니다. 이 접근법은 전통적인 익스플로잇 및 Groupby 조합의 복잡성을 피하고 특히 배열 내부의 미세한 변환 요소가 필요한 시나리오에 적합합니다. Spark에서 복잡한 반 구조화 된 데이터를 다룰 때 매우 유용한 트릭입니다.

위 내용은 Pyspark의 다층 중첩 어레이 구조를위한 평평한 기술의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!