Блог

Здесь у меня есть пара методов, которые занимают больше времени, чем хотелось бы. В настоящее время я упираюсь в стену, поскольку не вижу никакого очевидного способа написать эти методы более эффективным способом.

Для справки, то, что делает код, — это обработка набора данных о продажах, чтобы найти предыдущие заказы на продажу, связанные с тем же клиентом. Однако, как вы увидите, в середине много бизнес-логики, которая, вероятно, замедляет работу.

Я думал о рефакторинге этого в PySpark job, но прежде чем я это сделаю, я хотел бы знать, лучший ли это способ сделать это.

Я буду очень признателен за любые предложения здесь.

Больше контекста: выполнение каждого цикла занимает около 10 минут. В нем около 24 тыс. строк search_keys . Эти методы являются частью задачи Луиджи.

 def previous_commits(self, df: pd.DataFrame):
# Build some filters to slice data:
search_keys = df.loc[:, ['accountid', 'opportunityid']].drop_duplicates()
cols_a = ['opportunityid', 'opportunity_type', 'platform', 'closedate']
cols_b = ['opportunityid', 'productid']

# Build a list with the previous commit oppy_id:
commits = [
    {
        'acc_id': acc,
        'current_oppy': oppy,
        'previous_commit': self.fetch_latest_commit(oppy, df.loc[df.accountid == acc, cols_a].drop_duplicates())
    }
    for oppy, acc in tqdm(
        zip(search_keys.opportunityid, search_keys.accountid),
        desc='Finding previous commits data',
        file=sys.stdout,
        total=search_keys.shape[0]
    )
]

# Fetch products for the previous commit as well as the current row oppy:
products = [
    {
        'current_oppy': x.get('current_oppy'),
        'current_products': self.fetch_products_id(
            [x.get('current_oppy')],
            df.loc[df.accountid == x.get('acc_id'), cols_b].drop_duplicates()
        ),
        'previous_products': self.fetch_products_id(
            x.get('previous_commit'),
            df.loc[df.accountid == x.get('acc_id'), cols_b].drop_duplicates()
        ),
        'previous_recurrent_products': self.fetch_products_id(
            x.get('previous_commit'),
            df.loc[(df.accountid == x.get('acc_id')) amp; (df.fee_type == 'Recurring'), cols_b].drop_duplicates()
        )
    }
    for x in tqdm(
        commits,
        desc='Finding previous commit products',
        file=sys.stdout
    )
]

# Pick new calculated column and change its name for compatibility:
df = pd.DataFrame(commits).join(pd.DataFrame(products).set_index('current_oppy'), on='current_oppy')
df = df.loc[:, ['current_oppy', 'previous_commit', 'current_products', 'previous_recurrent_products']]
df.columns = ['current_oppy', 'previous_commit', 'current_products', 'previous_products']
return df

@staticmethod
def fetch_latest_commit(oppy_id: str, data: pd.DataFrame):
# Build some filters and create a df copy to search against:
data = data.set_index('opportunityid')
current_closedate = data.loc[data.index == oppy_id, ['closedate']].iat[0, 0]
current_platform = data.loc[data.index == oppy_id, ['platform']].iat[0, 0]
date_filter = data.closedate < current_closedate
platform_filter = data.platform == current_platform
eb_filter = data.opportunity_type != 'EB'
subset = data.loc[date_filter amp; eb_filter, :].sort_values('closedate', ascending=False)

if current_platform in {'CL', 'WE'}:
    # Fetch latest commit closedate for the same platform:
    subset = data.loc[date_filter amp; platform_filter amp; eb_filter, :].sort_values('closedate', ascending=False)
    latest_commit_date = subset.loc[:, 'closedate'].max()
    latest_commit_filter = subset.closedate == latest_commit_date
else:
    # Fetch latest commit closedate:
    latest_commit_date = subset.loc[:, 'closedate'].max()
    latest_commit_filter = subset.closedate == latest_commit_date

# Now try to get the latest commit oppy_id (if available), otherwise, just exit the function
# and return the current oppy_id. If the latest commit is a NB or NBU
# deal, then make another lookup to ensure that all the NB info is gathered since they might
# have different closedates.

try:
    latest_commit_id = list(subset.loc[latest_commit_filter, :].index)
    latest_commitid_filter = subset.index.isin(latest_commit_id)
    latest_commit_type = subset.loc[latest_commitid_filter, 'opportunity_type'].unique()[0]
except IndexError:
    return {oppy_id}

if latest_commit_type == 'RN':
    return set(latest_commit_id)
else:
    try:
        nb_before_latest_commit_filter = subset.closedate < latest_commit_date
        nb_only_filter = subset.opportunity_type == 'NB'
        nb_commit_id = list(subset.loc[nb_only_filter amp; nb_before_latest_commit_filter, :].index)
        return set(latest_commit_id   nb_commit_id)
    except IndexError:
        return set(latest_commit_id)

@staticmethod
def fetch_products_id(oppy_ids: list, data: pd.DataFrame):
    data = data.set_index('opportunityid')
    return set(data.loc[data.index.isin(oppy_ids), 'productid'])
 

«Очень простыми словами, Pandas выполняет операции на одной машине, в то время как PySpark работает на нескольких машинах. Если вы работаете над приложением машинного обучения, где имеете дело с большими наборами данных, лучше всего подходит PySpark, который может обрабатывать операции во много раз (в 100 раз) быстрее, чем Pandas. »

от https://sparkbyexamples.com/pyspark/pandas-vs-pyspark-dataframe-with-examples /

Вы также должны подумать о функциональном подходе Windows, чтобы получить предыдущий порядок. Это позволит избежать зацикливания на всех записях.