Создание Dict из CSV-потока данных

Question

Я пытаюсь сделать диктат из данных CSV в Python, я не хочу использовать традиционное разбиение (','), а затем использовать переименование строк в заголовок, который я хотел бы, так как я буду получать различные CSV файлы с разным объемом информации, и я не смогу последовательно ориентироваться на нужные мне строки этим методом.

ИМЕНА ЗАГОЛОВОК БУДЕТ ПОСТОЯННОЙ, просто их может быть больше заголовков в одном файле по сравнению с другим

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

Я могу создать список списков, используя csv.reader или:

class Split(beam.DoFn):
    def process(self, element):
        rows = element.splitlines()
        data = []
        for row in rows:
            data.append([row])
        return data

Возвращает:

[u'FIRST_NAME,last_name,birthdate,voter_id,phone_number']
[u'hector,ABAD,6/15/1970,11*******,7*********']
[u'm,ABAL,6/16/1949,12********,']
[u'jorge,ABDALA,6/15/1962,21********,3********']
[u'karen,ABELLA,6/18/1988,33********,']

Хотя, когда я пытаюсь получить доступ к первому ряду через:

rows = element.splitlines()
data = []
for row in rows:
    # f = pattern.findall(row)
    data.append([row])
return data[0]

Возвращает:

FIRST_NAME,last_name,birthdate,voter_id,phone_number
hector,ABAD,6/15/1970,11*******,7*********
m,ABAL,6/16/1949,109055849,
jorge,ABDALA,6/15/1962,21********,3********
karen,ABELLA,6/18/1988,33********,

Я также пробовал читатель beam_utils csv, хотя там говорится, что после исправления ошибки fileio не существует модуля с именем 'sources'.

Если кто-то знает лучший способ или может указать мне на то, что я делаю неправильно, это было бы замечательно, также это мой конвейер:

with beam.Pipeline(options=pipeline_options) as p:
    (p
     | 'Read' >> ReadFromText(known_args.input)
     | 'Split Values' >> beam.ParDo(Split())
     | 'WriteToText' >> beam.io.WriteToText(known_args.output)) 

Пока я только читаю из своего хранилища в облаке Google, но в будущем это будет из pubsub.

Я бы хотел, чтобы содержимое выглядело так:

{"FIRST_NAME": "hector", "last_name": "ABAD", "birthdate": "6/15/1970", "voter_id": 11*******, "phone_number": 7*********}
etc.
etc.
etc.

Answer 1

Обработка элемента заголовка CSV-файлов, похоже, не очень хорошо поддерживается Python beam SDK (кроме как его отбрасывание). К счастью, кто-то создал этот репозиторий для работы с этим вариантом использования: https://github.com/pabloem/beam_utils

Он содержит класс CSVFileSource, расширяющий FileBasedSource (абстрактный класс Beam для создания пользовательских файловых источников) для создания вашего dict из файла с переменными заголовками.

Установка:

pip install beam_utils
from beam_utils.sources import CsvFileSource

Может использоваться как:

 p | 'ReadCsvFile' >> beam.io.Read(CsvFileSource(known_args.input))

Должен выдать искомый результат.

Редактировать: чтобы сделать пакет доступным для работников Dataflow, создайте tar и предоставьте заданию флаг --extra_package, как в https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/#local-or-nonpypi

Answer 2

Проверьте модуль библиотеки Python csv.DictReader: https://docs.python.org/2/library/csv.html#csv.DictReader

Копирование примера из документации для быстрого ознакомления

>>> import csv
>>> with open('names.csv') as csvfile:
...     reader = csv.DictReader(csvfile)
...     for row in reader:
...         print(row['first_name'], row['last_name'])