Ошибка при использовании WriteToBigquery в python для конвейера потока данных.Объект Unicode не имеет атрибута 'items' - PullRequest
Новая
       9

Ошибка при использовании WriteToBigquery в python для конвейера потока данных.Объект Unicode не имеет атрибута 'items'

0 голосов
/ 25 ноября 2018

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

{
  "metadata": {
    "action": "insert",
    "type": "export",
    "version": 1,
    "timestamp": "2018-11-23T09:17:59.048-08:00"
  },
  "data": {
    "attr1": 61,
    "day": "2018-11-22",
    "pin": "2C49956",
    "CDP": 0,
    "DP": 0,
    "VD": 0,
    "seo": 0,
    "dir": 0,
    "other": 0,
    "at": 0
  }
}

Это простой файл, цель которого - запустить конвейер потока данных в пакетном режиме, чтобы вставить данные в таблицу больших запросов.В одном из преобразований, где я хочу взять метку времени из метаданных и добавить ее в качестве пары значения ключа в раздел данных, я получаю сообщение об ошибке из потока данных, в котором говорится, что «объект Unicode не имеет атрибута« items ».

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

import collections
import json
import argparse
import logging
from datetime import datetime
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions, SetupOptions, \
    WorkerOptions
from apache_beam.io.gcp import bigquery


# Creating options object
def create_options(argv):
    # pipeline options
    options = PipelineOptions()
    google_cloud_options = options.view_as(GoogleCloudOptions)
    google_cloud_options.project = 'something'
    google_cloud_options.job_name = datetime.now().strftime('somename')
    google_cloud_options.staging_location = 'some_loc'
    google_cloud_options.temp_location = 'another_loc'
    options.view_as(StandardOptions).runner = 'DirectRunner'
    options.view_as(SetupOptions).save_main_session = True
    options.view_as(WorkerOptions).machine_type = 'n1-standard-1'
    return options

    class PrepareData(beam.DoFn):
        """
        ParDo function to create a dictionary of data for downstream consumption
        """

        def process(self, element):
            data = json.loads(element)
            modified_data = {"action": data["metadata"]["action"], "timestamp": data["metadata"]["timestamp"], "data": data}
            return [modified_data]


    class FilterInserts(beam.DoFn):
        """
        Filter data for inserts
        """

        def process(self, element):
            if element["action"] == "insert":
                element['data']['data']['timestamp'] = element['timestamp']
                # for dict in element["data"]["data"]:
                #     dict["timestamp"] = element["timestamp"]
                return element["data"]["data"]


    def run_pipe(options, argv):
        """
        Creating pipelines
        """
        p = beam.Pipeline(options=options)

        main_pipe =p | 'PREPARE_DATA' >> beam.io.ReadFromText('/home/Downloads/sample_1') | beam.ParDo(PrepareData())

        """ Separating pipes for various actions """
        insert_pipe= main_pipe | beam.ParDo(FilterInserts())

        """
        Inserts--> sinking to BQ
        """
        insert_pipe | 'INSERT' >> beam.io.WriteToBigQuery(
            project='some-data-warehouse',
            dataset='sample_data',
            table='sample',
            write_disposition='WRITE_APPEND',
            create_disposition='CREATE_IF_NEEDED')


        p.run()


    def main():
        """
        Main function to drive the run
        :return: errors if any
        """
        parser = argparse.ArgumentParser()
        args = parser.parse_args()
        try:
            # create options
            opt = create_options(argv=args)
            # run pipeline
            run_pipe(opt, argv=args)
        except Exception as e:
            logging.error('Pipeline failed with error : %s', e)
            raise Exception('Pipeline failed with error : %s', e)


    if __name__ == "__main__":
        main()

Я запускаю его на прямом бегуне для тестирования на локальном компьютере, но я получаю ту же ошибку, даже если я изменяю бегун на dataflow-runner.Сообщение об ошибке: 

Exception: ('Pipeline failed with error : %s', AttributeError(u"'unicode' object has no attribute 'items' [while running 'INSERT/WriteToBigQuery']",))

Может кто-нибудь помочь мне выяснить, что идет не так и как я могу это исправить?

1 Ответ

0 голосов
/ 26 ноября 2018

Используя следующую таблицу schema (Вы можете изменить ее в соответствии со своими потребностями): 

schema = 'VD:INTEGER,pin:STRING,timestamp:STRING,other:INTEGER,CDP:INTEGER,dir:INTEGER,attr1:INTEGER,seo:INTEGER,day:STRING,DP:INTEGER,at:INTEGER'

Попробуйте выполнить следующие действия в классе FilterInserts: 

class FilterInserts(beam.DoFn):
    """
    Filter data for inserts
    """

    def process(self, element):
        if element["action"] == "insert":
            element['data']['data']['timestamp'] = element['timestamp']

            return [{
            'VD': element['data']['data']['VD'],
            'pin': element['data']['data']['pin'],
            'timestamp': element['data']['data']['timestamp'],
            'other': element['data']['data']['other'],
            'CDP': element['data']['data']['CDP'],
            'dir': element['data']['data']['dir'],
            'attr1' : element['data']['data']['attr1'],
            'seo' : element['data']['data']['seo'],
            'day' : element['data']['data']['day'],
            'DP' : element['data']['data']['DP'],
            'at' : element['data']['data']['at'],
            }]

Проблема возникает из-за того, что вам нужно отправить массив значений ключа в BigQuery, и вы отправляете словарь JSON со строками Unicode.

Надеюсь, это поможет.

...