Как работать с символами без ascii в BigQuery Stream Pipeline? - PullRequest
Новая
       19

Как работать с символами без ascii в BigQuery Stream Pipeline?

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

Как поддерживать не-ascii символы в конвейерах DataFlow, передаваемых в Google Cloud BigQuery в Python?

Этот код ниже пытается вставить одну строку в Big Query, но завершается неудачно с этой ошибкой:

ValidationError: Поле string_value обнаружило не-ASCII строку 'Этот код не выполнен из-за \ xc3 \ xa3, не являющегося символом ascii!': Кодек ascii не может декодировать байт 0xc3 в позиции 23: порядковый номер не находится в диапазоне(128)

Если мы удалим из текста символ «ã», код будет работать нормально. 

# -*- coding: utf-8 -*-

from __future__ import absolute_import
import argparse
import datetime

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions

options = PipelineOptions()
google_cloud_options = options.view_as(GoogleCloudOptions)


def run(argv=None):
    parser = argparse.ArgumentParser()
    known_args, pipeline_args = parser.parse_known_args(argv)
    p = beam.Pipeline(options=PipelineOptions(pipeline_args))

    (p
     | 'Create the PCollection' >> beam.Create([{
                'timestamp': datetime.datetime.now().isoformat(),
                'text': 'This code fail because of ã, a non-ascii character!',
            }])
     | 'Write to BigQuery' >> beam.io.Write(
                beam.io.BigQuerySink(
                    'lake.test',
                    schema='timestamp:TIMESTAMP,text:STRING',
                    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
                    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
                )
            )
     )

    p.run().wait_until_finish()


if __name__ == '__main__':
    run()
...