SQL или Python Самый эффективный способ построить последние отношения из истории событий

Question

У меня есть исторический набор данных, в котором контракты могут мигрировать в разные системы, и это может происходить бесконечно

Я открыт для решения этой проблемы с помощью Pyspark или SQL (в частности, красного смещения). SQL будет моим предпочтением. Я хочу отслеживать, какой был первый и последний контракт для каждой записи (на данный момент)

Набор данных выглядит следующим образом:

Old_contract, contract, migration_date
a123, b123, 2018-01-01
b123, c123, 2018-06-01
c123, d123, 2018-07-01
d123, e123, 2018-08-01
x123, y123, 2018-01-01
y123, z123, 2018-03-01
z123, y123, 2018-11-01

Вывод, который я бы искал: Обратите внимание, что второй пример приводит к y123, так как z123 был изменен обратно на y123

Old_contract, contract, migration_date, first_contract, last_contract
a123, b123, 2018-01-01, a123, e123
b123, c123, 2018-06-01, a123, e123
c123, d123, 2018-07-01, a123, e123
d123, e123, 2018-08-01, a123, e123
x123, y123, 2018-01-01, x123, y123
y123, z123, 2018-03-01, x123, y123
z123, y123, 2018-11-01, x123, y123

Answer 1

Это предполагает, что вся история контракта (его «след контракта», как я его назвал) происходит вместе в виде непрерывного ряда строк.Две соседние строки сгруппированы по contract из первой, что соответствует old_contract из второй.

Он написан как класс на случай, если вы захотите сделать что-нибудь более сложное позже.

Обратите внимание, что строка if not self.grouped(row, self): передает след контракта в качестве параметра функции группировки.Это означает, что вы можете сгруппировать новую строку на основе любой строки или строк уже в группе.В дальнейшем вам может понадобиться сгруппировать непоследовательные строки, используя несколько contract_trail объектов, возможно, используя некоторое время для завершения группы.

Приведенный ниже код достаточно эффективен и обрабатывает 100 000 строк / сек.на моем ноутбуке с Windows 10 i5.

import csv


class ContractTrail:
    def __init__(self, writer, grouped):
        self.writer = writer
        self.trail = []
        self.grouped = grouped

    def add(self, row):
        if not self.grouped(row, self):
            self.flush()
        self.trail.append(row)

    def flush(self):
        rows = self.rewrite_transactions()
        if rows:
            self.writer.writerows(rows)
        self.trail = []

    def rewrite_transactions(self):
        if self.trail:
            first_contract = self.trail[0]["old_contract"]
            last_contract = self.trail[-1]["contract"]
            for transaction in self.trail:
                transaction["first_contract"] = first_contract
                transaction["last_contract"] = last_contract
        return self.trail

    def flush_to_csv(self, writer):
        rows = self.rewrite_transactions()
        if rows:
            writer.writerows(rows)


def link_to_previous_row(current_row, contract_trail):
    if contract_trail.trail:
        return current_row["old_contract"] == contract_trail.trail[-1]["contract"]
    # empty trail implies just started-- include current row in the group
    return True


if __name__ == '__main__':
    with open('trans.csv') as infile:
        with open('trans_out.csv', 'w', newline='\n') as outfile:
            reader = csv.DictReader(infile)
            write_fieldnames = reader.fieldnames + ['first_contract', 'last_contract']
            writer = csv.DictWriter(outfile, fieldnames=write_fieldnames)
            writer.writeheader()
            contract_trail = ContractTrail(grouped=link_to_previous_row, writer=writer)
            for row in reader:
                contract_trail.add(row)
            contract_trail.flush()

>>> python transaction_history.py

trans_out.csv:
old_contract,contract,migration_date,first_contract,last_contract
a123,b123,2018-01-01,a123,e123
b123,c123,2018-06-01,a123,e123
c123,d123,2018-07-01,a123,e123
d123,e123,2018-08-01,a123,e123
x123,y123,2018-01-01,x123,y123
y123,z123,2018-03-01,x123,y123
z123,y123,2018-11-01,x123,y123