искра присоединиться к различным уровням соответствия - PullRequest
Новая
       27

искра присоединиться к различным уровням соответствия

0 голосов
/ 06 декабря 2018

У меня есть два искровых фрейма данных: 

df1 = sc.parallelize([
    ['a', '1', 'value1'],
    ['b', '1', 'value2'],
    ['c', '2', 'value3'],
    ['d', '4', 'value4'],
    ['e', '2', 'value5'],
    ['f', '4', 'value6']
]).toDF(('id1', 'id2', 'v1'))

df2 = sc.parallelize([
    ['a','1', 1],
    ['b','1', 1],
    ['y','2', 4],
    ['z','2', 4]
]).toDF(('id1', 'id2', 'v2'))

Каждый из них имеет поля id1 и id2 (и может содержать много идентификаторов).Сначала мне нужно сопоставить df1 с df2 на id1 .Затем мне нужно сопоставить все несопоставленных записей с обоих кадров данных по id2 и т. Д.

Мой путь: 

def joinA(df1,df2, field):
    from pyspark.sql.functions import lit

    L = 'L_'
    R = 'R_'
    Lfield = L+field
    Rfield = R+field

    # Taking field's names
    df1n = df1.schema.names
    df2n = df2.schema.names
    newL = [L+fld for fld in df1n]
    newR = [R+fld for fld in df2n]

    # drop duplicates by input field
    df1 = df1.toDF(*newL).dropDuplicates([Lfield])
    df2 = df2.toDF(*newR).dropDuplicates([Rfield])

    # matching records
    df_full = df1.join(df2,df1[Lfield]==df2[Rfield],how = 'outer').cache()

    # unmatched records from df1
    df_left = df_full.where(df2[Rfield].isNull()).select(newL).toDF(*df1n)
    # unmatched records from df2
    df_right = df_full.where(df1[Lfield].isNull()).select(newR).toDF(*df2n)
    # matched records and adding match level
    df_inner = df_full.where(\
        (~df1[Lfield].isNull()) & (~df2[Rfield].isNull())\
    ).withColumn('matchlevel',lit(field))

    return df_left, df_inner, df_right


first_l,first_i,first_r = joinA(df1,df2,'id1')
second_l,second_i,second_r = joinA(first_l,first_r,'id2')

result = first_i.union(second_i)

Есть ли способ сделать это проще?Или какие-то стандартные инструменты для этой работы?

Спасибо,

Макс

1 Ответ

0 голосов
/ 06 декабря 2018

У меня есть другой способ сделать это ... но я не уверен, что это лучше, чем ваше решение: 

from pyspark.sql import functions as F

id_cols = [cols for cols in df1.columns if cols != 'v1']

df1 = df1.withColumn("get_v2", F.lit(None))
df1 = df1.withColumn("match_level", F.lit(None))

for col in id_cols:
    new_df1 = df1.join(
        df2.select(
            col, 
            "v2"
        ),
        on=(
            (df1[col] == df2[col])
            & df1['get_v2'].isNull()
        ),
        how='left'
    )

    new_df1 = new_df1.withColumn(
        "get_v2",
        F.coalesce(df1.get_v2, df2.v2)
    ).drop(df2[col]).drop(df2.v2)

    new_df1 = new_df1.withColumn(
      "match_level",
      F.when(F.col("get_v2").isNotNull(), F.coalesce(F.col("match_level"), F.lit(col)))
    )

    df1 = new_df1

df1.show()
+---+---+---+------+------+-----------+
|id1|id2|id3|    v1|get_v2|match_level|
+---+---+---+------+------+-----------+
|  f|  4|  1|value6|     3|        id3|
|  d|  4|  1|value4|     3|        id3|
|  c|  2|  1|value3|     4|        id2|
|  c|  2|  1|value3|     4|        id2|
|  e|  2|  1|value5|     4|        id2|
|  e|  2|  1|value5|     4|        id2|
|  b|  1|  1|value2|     1|        id1|
|  a|  1|  1|value1|     1|        id1|
+---+---+---+------+------+-----------+

это приведет к N-соединениям, где N - это число идентификаторов, которые вы получили.

РЕДАКТИРОВАТЬ : добавлен уровень совпадения!

...