Как я могу остановить дополнительное повторение в доходности / доходности, сохраняя при этом промежуточные итоги для данной пары ключ: значение? - PullRequest
Новая
       11

Как я могу остановить дополнительное повторение в доходности / доходности, сохраняя при этом промежуточные итоги для данной пары ключ: значение?

0 голосов
/ 01 апреля 2019

После передачи Pcollection к следующему преобразованию доходность / доходность преобразования умножается, когда мне нужна только одна пара KV для данной улицы и количества аварий.

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

class CountAccidents(beam.DoFn):
    acci_dict = {}

    def process(self, element):
        if self.acci_dict.__contains__(element[0]['STREET_NAME']):
            self.acci_dict[element[0]['STREET_NAME']] += 1
        else:
            self.acci_dict.update({element[0]['STREET_NAME']: 1})
        if self.acci_dict != {}:
            yield self.acci_dict


def run():
    with beam.Pipeline() as pl:
        test = (pl | 'Read' >> beam.io.ReadFromText('/modified_Excel_Crashes_Chicago.csv')
                   | 'Map Accident' >> beam.ParDo(AccidentstoDict())
                   | 'Count Accidents' >> beam.ParDo(CountAccidents())
                   | 'Print to Text' >> beam.io.WriteToText('/letstestthis', file_name_suffix='.txt'))```                                                      

Input Pcollection:
[{'CRASH_DATE': '3/25/19 0:25', 'WEATHER_CONDITION': 'CLEAR', 'STREET_NAME': 'KOSTNER AVE', 'CRASH_HOUR': '0'}]
[{'CRASH_DATE': '3/24/19 23:40', 'WEATHER_CONDITION': 'CLEAR', 'STREET_NAME': 'ARCHER AVE', 'CRASH_HOUR': '23'}]
[{'CRASH_DATE': '3/24/19 23:30', 'WEATHER_CONDITION': 'UNKNOWN', 'STREET_NAME': 'VAN BUREN ST', 'CRASH_HOUR': '23'}]

I expect to get this: 
{'KILPATRICK AVE': 1, 'MILWAUKEE AVE': 1, 'CENTRAL AVE': 2, 'WESTERN AVE': 6, 'DANTE AVE': 1}

What I get is this(a slow build-up till complete): 
{'KOSTNER AVE': 1}
{'KOSTNER AVE': 1, 'ARCHER AVE': 1}
{'KOSTNER AVE': 2, 'ARCHER AVE': 2, 'VAN BUREN ST': 1}

1 Ответ

0 голосов
/ 01 апреля 2019

Вам потребуется выполнить объединение для каждого ключа, для Count вы можете использовать его здесь:

https://beam.apache.org/releases/pydoc/2.9.0/apache_beam.transforms.combiners.html

После операции чтения выведите KeyValue, которое равно {STREET, 1}, за которым следует преобразование «Количество на ключ», которое даст вам общее количество улиц.

Оттуда было бы легко добавить функции управления окнами, если вы хотите, например, вывод в неделю.Вам просто нужно добавить метку времени и окно в вызове.Пример того, как это сделать, приведен здесь:

Как назначить временные метки для данных из пакетных источников, например, для файлов CSV в конвейере Beam

...