Невозможно преобразовать набор лучей Python в список

Question

TypeError: 'PCollection' object does not support indexing

Приведенная выше ошибка возникает в результате попытки преобразования Pcollection в список:

filesList = (files | beam.combiners.ToList())

lines = (p | 'read' >> beam.Create(ReadSHP().ReadSHP(filesList))
            | 'map' >> beam.Map(_to_dictionary))

И:

def ReadSHP(self, filesList):
    """
    """
    sf = shp.Reader(shp=filesList[1], dbf=filesList[2])  

Как решить эту проблему?Любая помощь приветствуется.

Answer 1

Альтернативным вариантом будет использование виртуальной машины GCE и преобразование шейп-файлов в GeoJSON с использованием таких инструментов, как ogr2ogr.Затем GeoJSON может быть загружен в BigQuery и может быть запрошен с помощью BigQuery GIS.

Вот блог-пост с более подробной информацией
https://medium.com/google-cloud/how-to-load-geographic-data-like-zipcode-boundaries-into-bigquery-25e4be4391c8

Answer 2

Как правило, PCollection нельзя преобразовать в список.

PCollection - это набор элементов, который потенциально неограничен и неупорядочен.Луч позволяет применять преобразования к PCollection.Применение PTransform к PCollection дает еще один PCollection.И процесс применения преобразования потенциально распределяется на парк машин.Поэтому в общем случае невозможно преобразовать такую ​​вещь в набор элементов в локальной памяти.

Комбинаторы - это просто специальный класс PTransforms.Они накапливают все элементы, которые видят, применяют к элементам некоторую логику объединения, а затем выводят результат объединения.Например, объединитель может просмотреть входящие элементы, суммировать их, а затем вывести сумму в результате.Такой объединитель преобразует PCollection элементов в PCollection сумм этих элементов.

beam.combiners.ToList - это просто еще одно преобразование, которое применяется к PCollection, возможно, по парку рабочих машин,и дает еще один PCollection.Но на самом деле он не выполняет сложного объединения до получения выходных элементов, он только накапливает все видимые элементы в списке, а затем выводит список видимых элементов.Таким образом, он берет элементы, которые являются парами ключ-значение (на нескольких машинах), помещает их в списки и выводит эти списки.

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

Один из обходных путей - добавить еще один шаг в конвейер, который записываетобъединенные выходные данные (например, суммы или списки) в общее хранилище, например таблицу в некоторой базе данных или файл.И затем, когда конвейер завершает свою работу, программа может загрузить результаты выполнения конвейера из этого места.

Подробнее см. В документации:

• Модель выполнения луча: https://beam.apache.org/documentation/execution-model/
• Проектирование вашего трубопровода: https://beam.apache.org/documentation/pipelines/design-your-pipeline/
• Руководство по программированию луча: https://beam.apache.org/documentation/programming-guide/