Spark는 열의 배열에서 동일한 값을 가진 모든 행을 가져옵니다.

Spark 3에서 다음 코드는 두 배열이 동일한 위치에서 하나 이상의 요소를 공유하는 행만 유지하면서 행 간의 배열을 비교합니다. df입력 데이터 프레임입니다.

    df.join(
      df.withColumnRenamed("id", "id2").withColumnRenamed("hashes", "hashes2"),
      exists(arrays_zip(col("hashes"), col("hashes2")), x => x("hashes") === x("hashes2"))
    )
      .groupBy("id")
      .agg(first(col("hashes")).as("hashes"), collect_list("id2").as("matched"))
      .withColumn("matched", filter(col("matched"), x => x.notEqual(col("id"))))

상세 설명

먼저 두 개의 해시 배열에서 동일한 위치에있는 최소 하나의 요소 조건에 따라 필터링 된 자동 교차 조인을 수행합니다.

조건을 구축하기 위해 첫 번째 데이터 프레임에서 하나씩, 두 번째 결합 된 데이터 프레임에 대해 하나씩, 즉 열 이름이 변경된 첫 번째 데이터 프레임 인 두 개의 해시 배열을 압축합니다. 압축하여 배열을 얻고 {"hashes":x, "hashes2":y}다음으로이 배열에 요소가 존재하는지 확인해야합니다 x = y. 완전한 조건은 다음과 같이 작성됩니다.

exists(arrays_zip(col("hashes"), col("hashes2")), x => x("hashes") === x("hashes2"))

그런 다음 열별로 집계하여 보관 된 id모든 id2행, 즉 조건과 일치하는 행 을 수집합니다.

"해시"열을 유지하기 위해 동일한 "id"를 가진 두 행의 "해시"열이 같으면 각 "id"에 대해 "해시"가 처음으로 나타납니다. 그리고 collect_list를 사용하여 모든 "id2"를 수집합니다 .

.agg(first(col("hashes")).as("hashes"), collect_list("id2").as("matches"))

마지막으로 현재 행의 ID와 "일치하는"열에서 필터링합니다.

.withColumn("matches", filter(col("matches"), x => x.notEqual(col("id"))))

순서대로 "id"가 필요한 경우 orderBy절을 추가 할 수 있습니다 .

.orderBy("id")

운영

df다음 값을 포함 하는 데이터 프레임 사용 :

+---+---------------+
|id |hashes         |
+---+---------------+
|0  |[1, 2, 3, 4, 5]|
|1  |[1, 5, 3, 7, 9]|
|2  |[9, 3, 6, 8, 0]|
+---+---------------+

다음 출력이 표시됩니다.

+---+---------------+-------+
|id |hashes         |matches|
+---+---------------+-------+
|0  |[1, 2, 3, 4, 5]|[1]    |
|1  |[1, 5, 3, 7, 9]|[0]    |
|2  |[9, 3, 6, 8, 0]|[]     |
+---+---------------+-------+

제한

조인은 데카르트 곱으로 매우 비쌉니다. 조건이 결과를 필터링하더라도 큰 데이터 세트에서 엄청난 양의 계산 / 셔플이 발생할 수 있으며 성능이 매우 저하 될 수 있습니다.

3.0 이전 버전의 Spark를 사용하는 경우 일부 내장 Spark 함수 를 사용자 정의 함수 로 대체해야 합니다.