Сравнение двух строк одновременно в PySpark - PullRequest
Новая
       21

Сравнение двух строк одновременно в PySpark

0 голосов
/ 09 мая 2018

Я новичок в Spark и ищу помощи с лучшими практиками. У меня большой DataFrame, и мне нужно передать две строки за раз в функцию, которая сравнивает их.

actual_data - это DataFrame со столбцом id и несколькими столбцами значений.

rows_to_compare - это DataFrame с двумя столбцами: left_id и right_id.

Для каждой пары в rows_to_compare я бы хотел передать две соответствующие строки из actual_data в функцию.

Мои фактические данные довольно большие (~ 30 ГБ) и содержат много столбцов, поэтому я сократил их до следующего более простого примера: 

import pandas as pd
from pyspark.sql import SQLContext
from pyspark.sql.functions import col
import builtins

sqlContext = SQLContext(sc)
# Build DataFrame of Actual Data
data = { 
    'id': [1,2,3,4,5],
    'value': [11,12,13,14,15]}
actual_data_df = sqlContext.createDataFrame(
    pd.DataFrame(data, columns=data.keys()))

# Build DataFrame of Rows To Compare
rows_to_compare = {
    'left_id': [1,2,3,4,5],
    'right_id': [1,1,1,1,1]}
rows_to_compare_df = 
    sqlContext.createDataFrame(
        pd.DataFrame(rows_to_compare, columns=rows_to_compare.keys()))

result = (
    rows_to_compare_df
        .join(
            actual_data_df.alias('a'),
            col('left_id') == col('a.id'))
        .join(
            actual_data_df.alias('b'),
            col('right_id') == col('b.id'))
        .withColumn(
            'total',
            builtins.sum(
                [col('a.value'), 
                 col('b.value')]))
        .select('a.id', 'b.id', 'total')
        .collect())

Возвращает желаемый результат:

[Row(id=2, id=1, total=23), Row(id=5, id=1, total=26), Row(id=4, id=1, total=25), Row(id=1, id=1, total=22), Row(id=3, id=1, total=24)]

Когда я запускаю это, оно кажется довольно медленным, даже для этой проблемы с игрушкой. Это лучший способ решения этой проблемы? Самый ясный альтернативный подход, который я могу придумать, состоит в том, чтобы каждая строка моего DataFrame содержала значения для обеих строк, которые я хотел бы сравнить. Я обеспокоен этим подходом, так как он предполагает огромное количество дублирования данных.

Любая помощь очень ценится, спасибо.

...