Как развернуть Google Cloud Dataflow с подключением к PostgreSQL (лучи-самородки) из Google Cloud Functions - PullRequest
Новая
       86

Как развернуть Google Cloud Dataflow с подключением к PostgreSQL (лучи-самородки) из Google Cloud Functions

1 голос
/ 13 июля 2020

Я пытаюсь создать ETL в GCP, который будет читать часть данных из PostgreSQL и помещать их в подходящей форме в BigQuery. Мне удалось выполнить эту задачу, развернув поток данных со своего компьютера, но мне не удалось сделать его динамическим c, поэтому он будет читать последнюю переданную запись и передавать следующие 100. Итак, я понял, что буду создавать потоки данных из облачной функции . Все работало нормально, чтение / запись в BigQuery работает как шарм, но я застрял на PostgreSQL Requited package: beam-nuggets.

В функции я создаю аргументы канала: 

pipe_arguments = [    
    '--project={0}'.format(PROJECT),
    '--staging_location=gs://xxx.appspot.com/staging/',
    '--temp_location=gs://xxx.appspot.com/temp/',
    '--runner=DataflowRunner',
    '--region=europe-west4',
    '--setup_file=./setup.py'
    ]

    pipeline_options = PipelineOptions(pipe_arguments)
    pipeline_options.view_as(SetupOptions).save_main_session = save_main_session

Затем создайте конвейер: 

 pipeline = beam.Pipeline(argv = pipe_arguments)

и запустите его: 

pipeline.run()

Если я опущу: 

    '--setup_file=./setup.py'

все в порядке, кроме потока данных невозможно использовать PostgeQSL в качестве импорта: 

from beam_nuggets.io import relational_db

не удается.

Когда я добавляю строку 

    '--setup_file=./setup.py'

, функция тестирования с веб-портала GCP Function возвращает: 

Error: function terminated. Recommended action: inspect logs for termination reason. Details:
Full trace: Traceback (most recent call last):
  File "/env/local/lib/python3.7/site-packages/apache_beam/utils/processes.py", line 85, in check_output
    out = subprocess.check_output(*args, **kwargs)
  File "/opt/python3.7/lib/python3.7/subprocess.py", line 411, in check_output
    **kwargs).stdout
  File "/opt/python3.7/lib/python3.7/subprocess.py", line 512, in run
    output=stdout, stderr=stderr)
subprocess.CalledProcessError: Command '['/env/bin/python3.7', 'setup.py', 'sdist', '--dist-dir', '/tmp/tmpxdvj0ulx']' returned non-zero exit status 1.
,          output of the failed child process b'running sdist\nrunning egg_info\ncreating example.egg-info\n'

работает 

python setup.py sdist --dist-dir ./tmp/

с локального компьютера работает нормально.

setup.py развертывается вместе с кодом функции (main.py) и requirements.txt для облачной функции.

Requirements.txt используется во время развертывания функции и выглядит так: 

beam-nuggets==0.15.1
google-cloud-bigquery==1.17.1
apache-beam==2.19.0
google-cloud-dataflow==2.4.0
google-apitools==0.5.31

setup.py выглядит так: 

from setuptools import find_packages
from setuptools import setup

REQUIRED_PACKAGES = ['beam-nuggets>=0.15.1']

setup(
    name='example',
    version='0.1',
    install_requires=REQUIRED_PACKAGES,
    packages=find_packages(),
    include_package_data=True,
    description='example desc'
)

Я застрял на пару дней, пробовал разные подходы setup.py, пытался использовать requirements.txt вместо setup.py - не повезло.

журнал просто говорит:

* 104 2 *

Дополнительная информация:

если я использую 

'--requirements_file=./requirements.txt'

вместо 

'--setup_file=./setup.py'

Я получаю: 

Error: memory limit exceeded.

на веб-портале GCP Functions при запуске функции тестирования.

После того, как я увеличил объем памяти до 2 ГБ, он говорит: 

Error: function terminated. Recommended action: inspect logs for termination reason. Details:
Full traceback: Traceback (most recent call last):
  File "/env/local/lib/python3.7/site-packages/apache_beam/utils/processes.py", line 85, in check_output
    out = subprocess.check_output(*args, **kwargs)
  File "/opt/python3.7/lib/python3.7/subprocess.py", line 411, in check_output
    **kwargs).stdout
  File "/opt/python3.7/lib/python3.7/subprocess.py", line 512, in run
    output=stdout, stderr=stderr)
subprocess.CalledProcessError: Command '['/env/bin/python3.7', '-m', 'pip', 'download', '--dest', '/tmp/dataflow-requirements-cache', '-r', './requirements.txt', '--exists-action', 'i', '--no-binary', ':all:']' returned non-zero exit status 1. 
 Pip install failed for package: -r         
 Output from execution of subprocess: b'Collecting beam-nuggets==0.15.1  
 Downloading beam-nuggets-0.15.1.tar.gz (17 kB)
  Saved /tmp/dataflow-requirements-cache/beam-nuggets-0.15.1.tar.gz
Collecting google-cloud-bigquery==1.17.1
  Downloading google-cloud-bigquery-1.17.1.tar.gz (228 kB)
  Saved /tmp/dataflow-requirements-cache/google-cloud-bigquery-1.17.1.tar.gz
Collecting apache-beam==2.19.0
  Downloading apache-beam-2.19.0.zip (1.9 MB)
  Saved /tmp/dataflow-requirements-cache/apache-beam-2.19.0.zip
Collecting google-cloud-dataflow==2.4.0
  Downloading google-cloud-dataflow-2.4.0.tar.gz (5.8 kB)
  Saved /tmp/dataflow-requirements-cache/google-cloud-dataflow-2.4.0.tar.gz
Collecting google-apitools==0.5.31
  Downloading google-apitools-0.5.31.tar.gz (173 kB)
  Saved /tmp/dataflow-requirements-cache/google-apitools-0.5.31.tar.gz
Collecting SQLAlchemy<2.0.0,>=1.2.14
  Downloading SQLAlchemy-1.3.18.tar.gz (6.0 MB)
  Saved /tmp/dataflow-requirements-cache/SQLAlchemy-1.3.18.tar.gz
Collecting sqlalchemy-utils<0.34,>=0.33.11
  Downloading SQLAlchemy-Utils-0.33.11.tar.gz (128 kB)
  Saved /tmp/dataflow-requirements-cache/SQLAlchemy-Utils-0.33.11.tar.gz
Collecting pg8000<2.0.0,>=1.12.4
  Downloading pg8000-1.16.0.tar.gz (75 kB)
  Saved /tmp/dataflow-requirements-cache/pg8000-1.16.0.tar.gz
Collecting PyMySQL<2.0.0,>=0.9.3
  Downloading PyMySQL-0.9.3.tar.gz (75 kB)
  Saved /tmp/dataflow-requirements-cache/PyMySQL-0.9.3.tar.gz
Collecting kafka>===1.3.5
  Downloading kafka-1.3.5.tar.gz (227 kB)
  Saved /tmp/dataflow-requirements-cache/kafka-1.3.5.tar.gz
Collecting google-cloud-core<2.0dev,>=1.0.0
 Downloading google-cloud-core-1.3.0.tar.gz (32 kB)
  Saved /tmp/dataflow-requirements-cache/google-cloud-core-1.3.0.tar.gz
Collecting google-resumable-media<0.5.0dev,>=0.3.1
  Downloading google-resumable-media-0.4.1.tar.gz (2.1 MB)
  Saved /tmp/dataflow-requirements-cache/google-resumable-media-0.4.1.tar.gz
Collecting protobuf>=3.6.0
  Downloading protobuf-3.12.2.tar.gz (265 kB)
  Saved /tmp/dataflow-requirements-cache/protobuf-3.12.2.tar.gz
Collecting crcmod<2.0,>=1.7
  Downloading crcmod-1.7.tar.gz (89 kB)
  Saved /tmp/dataflow-requirements-cache/crcmod-1.7.tar.gz
Collecting dill<0.3.2,>=0.3.1.1
  Downloading dill-0.3.1.1.tar.gz (151 kB)
  Saved /tmp/dataflow-requirements-cache/dill-0.3.1.1.tar.gz
Collecting fastavro<0.22,>=0.21.4
  Downloading fastavro-0.21.24.tar.gz (496 kB)
  Saved /tmp/dataflow-requirements-cache/fastavro-0.21.24.tar.gz
Collecting future<1.0.0,>=0.16.0
  Downloading future-0.18.2.tar.gz (829 kB)
  Saved /tmp/dataflow-requirements-cache/future-0.18.2.tar.gz
Collecting grpcio<2,>=1.12.1
  Downloading grpcio-1.30.0.tar.gz (19.7 MB)
    ERROR: Command errored out with exit status 1:
     command: /env/bin/python3.7 -c \'import sys, setuptools, tokenize; sys.argv[0] = \'"\'"\'/tmp/pip-download-yjpzrbur/grpcio/setup.py\'"\'"\'; __file__=\'"\'"\'/tmp/pip-download-yjpzrbur/grpcio/setup.py\'"\'"\';f=getattr(tokenize, \'"\'"\'open\'"\'"\', open)(__file__);code=f.read().replace(\'"\'"\'\\r\
\'"\'"\', \'"\'"\'\
\'"\'"\');f.close();exec(compile(code, __file__, \'"\'"\'exec\'"\'"\'))\' egg_info --egg-base /tmp/pip-download-yjpzrbur/grpcio/pip-egg-info
         cwd: /tmp/pip-download-yjpzrbur/grpcio/
    Complete output (11 lines):
    Traceback (most recent call last):
      File "<string>", line 1, in <module>
      File "/tmp/pip-download-yjpzrbur/grpcio/setup.py", line 196, in <module>
        if check_linker_need_libatomic():
      File "/tmp/pip-download-yjpzrbur/grpcio/setup.py", line 156, in check_linker_need_libatomic
        stderr=PIPE)
      File "/opt/python3.7/lib/python3.7/subprocess.py", line 800, in __init__
        restore_signals, start_new_session)
      File "/opt/python3.7/lib/python3.7/subprocess.py", line 1551, in _execute_child
        raise child_exception_type(errno_num, err_msg, err_filename)
    FileNotFoundError: [Errno 2] No such file or directory: \'cc\': \'cc\'
    ----------------------------------------
ERROR: Command errored out with exit status 1: python setup.py egg_info Check the logs for full command output.
WARNING: You are using pip version 20.0.2; however, version 20.1.1 is available.
You should consider upgrading via the \'/env/bin/python3.7 -m pip install --upgrade pip\' command.
'

Журналы в этом случае: 

 {
 insertId: "000000-5e4c10f4-d542-4631-8aaa-b9306d1390fd"  
 
labels: {
  execution_id: "15jww0sd8uyz"   
 }
 logName: "projects/xxx/logs/cloudfunctions.googleapis.com%2Fcloud-functions"  
 receiveTimestamp: "2020-07-13T14:01:33.505683371Z"  
 
resource: {
  
labels: {
   function_name: xxx"    
   project_id: "xxx"    
   region: "europe-west6"    
  }
  type: "cloud_function"   
 }
 severity: "DEBUG"  
 textPayload: "Function execution took 18984 ms, finished with status: 'crash'"  
 timestamp: "2020-07-13T14:01:32.953194652Z"  
 trace: "projects/xxx/traces/262224a3d230cd9a66b1eebba3d7c3e0"  
}

Развертывание потока данных с локального компьютера работает нормально.

Команда из журналов: 

python -m pip download --dest ./tmp -r ./requirements.txt --exists-action i --no-binary :all:

также работает нормально, хотя кажется, что загружается половина inte rnet на пару минут, даже если Я уменьшаю файл requirements.txt только до beam-nuggets == 0.15.1.

Он застревает на 

grpcio-1.30.0.tar.gz (19.7 MB)

именно во время установки из этого пакета, функция: 

def check_linker_need_libatomic():
    """Test if linker on system needs libatomic."""
    code_test = (b'#include <atomic>\n' +
                 b'int main() { return std::atomic<int64_t>{}; }')
    cc_test = subprocess.Popen(['cc', '-x', 'c++', '-std=c++11', '-'],
                               stdin=PIPE,
                               stdout=PIPE,
                               stderr=PIPE)
    cc_test.communicate(input=code_test)
    return cc_test.returncode != 0

1 Ответ

1 голос
/ 15 июля 2020

Я также попробовал GCP AppEngine вместо Cloud Functions, с тем же результатом, однако он направляет меня к правильному решению. Благодаря this и this я смог создать внешний пакет из самородков лучей и включить его, используя --extra_package вместо --setup_file или --setup_file.

Проблема с компиляцией grpcio (вызванная ненастраиваемым --no-binary', ':all:') остается. Проблема со странной ошибкой setup.py также остается.

Но развертывание из облачных функций в поток данных (с зависимостями) работает, поэтому проблема закрыта для меня.

Обновление:

Сразу после этого у меня возникла проблема: 

in _import_module return __import__(import_name) ModuleNotFoundError: No module named 'main'

, поскольку я не использовал какой-либо «основной» модуль, который было трудно найти, я должен упаковать во внешний пакет также каждую функцию, определенную в мой файл main.py (отсюда и имя модуля). Итак, файл extra_package содержит все внешние зависимости и мой собственный модуль, в котором хранятся мои функции.

...