ModuleNotFoundError：运行 GCP Dataflow 作业时没有名为 \'oracledb\' 的模块答案

【问题标题】：ModuleNotFoundError: No module named 'oracledb' when running GCP Dataflow jobsModuleNotFoundError：运行 GCP Dataflow 作业时没有名为 \'oracledb\' 的模块
【发布时间】：2022-08-19 04:44:34
【问题描述】：

我们正在尝试使用 GCP 数据流和 Python 作业模板连接到 Oracle 数据库。由于我们使用无法访问 Internet 的特殊子网来运行 Dataflow 作业，因此我们使用 setup.py 从 GCS 存储桶安装依赖包。

下面是使用 setup.py 创建数据流模板的命令行：

python3 -m <python_file_name> --runner DataflowRunner --project <project_id> --staging_location <gcs_staging> --temp_location <gcs_temp> --template_location <gcs_template> --region <region> --setup_file=./setup.py

依赖包存储在 GCP 存储桶中，并将在作业运行时复制到 Dataflow 工作器并安装在 Dataflow 工作器上。 Oracle数据库连接，我们使用oracledb-1.0.3-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl，下载自https://pypi.org/project/oracledb/#files。

当我们尝试使用 Cloud Shell 和 DirectRunner 时，它可以成功安装并识别 oracledb 模块。但是，当 Dataflow 作业执行时，它会遇到以下错误：

来自工作人员的错误消息：回溯（最近一次调用最后一次）：文件 \"/usr/local/lib/python3.9/site-packages/dataflow_worker/batchworker.py\"，第 772 行，在运行 self._load_main_session(self. local_staging_directory) 文件 \"/usr/local/lib/python3.9/site-packages/dataflow_worker/batchworker.py\"，第 509 行，在 _load_main_session pickler.load_session(session_file) 文件 \"/usr/local/lib/python3 .9/site-packages/apache_beam/internal/pickler.py\"，第 65 行，在 load_session 中返回 desired_pickle_lib.load_session(file_path) 文件 \"/usr/local/lib/python3.9/site-packages/apache_beam/internal /dill_pickler.py\"，第 313 行，在 load_session 中返回 dill.load_session(file_path) 文件 \"/usr/local/lib/python3.9/site-packages/dill/_dill.py\"，第 368 行，在 load_session模块 = unpickler.load() 文件 \"/usr/local/lib/python3.9/site-packages/dill/_dill.py\"，第 472 行，在加载 obj = StockUnpickler.load(self) 文件 \"/ usr/local/lib/python3.9/site-packages/dill/_dill.py\"，第 826 行，在 _import_module 返回进口(import_name) ModuleNotFoundError: 没有名为 \'oracledb\' 的模块

非常感谢您的建议。

安装程序.py

import os
import logging
import subprocess
import pickle
import setuptools
import distutils

from setuptools.command.install import install as _install

class install(_install):  # pylint: disable=invalid-name
    def run(self):
        self.run_command(\'CustomCommands\')
        _install.run(self)


WHEEL_PACKAGES = [
    \'wheel-0.37.1-py2.py3-none-any.whl\',
    \'oracledb-1.0.3-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl\'
    ]
CUSTOM_COMMANDS = [
    [\'sudo\', \'apt-get\', \'update\']
]

class CustomCommands(setuptools.Command):
    \"\"\"A setuptools Command class able to run arbitrary commands.\"\"\"

    def initialize_options(self):
        pass

    def finalize_options(self):
        pass

    def run_command(self, command):
        import subprocess
        import logging
        logging.getLogger().setLevel(logging.INFO) 
        status = -9999
        try:
            logging.info(\'CUSTOM_DATAFLOW_JOB_LOG: started running [{}]\'.format(command))
            status = subprocess.call(command)
            if status == 0:
                logging.info(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] completed successfully\'.format(command))
            else:
                logging.error(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] failed with signal {}\'.format(command, status))
        except Exception as e:
            logging.error(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] caught exception: {}\'.format(command, e))
        return status        

    def install_cmd(self):
        result = []
        for p in WHEEL_PACKAGES:
                result.append([\'gsutil\', \'cp\', \'gs://dataflow-execution/python_dependencies/{}\'.format(p), \'.\'])
                result.append([\'pip\', \'install\', \'{}\'.format(p)])
        return result

    def run(self):
        import logging
        logging.getLogger().setLevel(logging.INFO) 
        try:
            install_cmd = self.install_cmd()
            for command in CUSTOM_COMMANDS:
                status = self.run_command(command)
                if status == 0:
                    logging.info(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] finished successfully\'.format(command))
                else:
                    logging.error(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] failed with status code {}\'.format(command, status))
            for command in install_cmd:
                status = self.run_command(command)
                if status == 0:
                    logging.info(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] finished successfully\'.format(command))
                else:
                    logging.error(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] failed with status code {}\'.format(command, status))
        except Exception as e:
            logging.error(\'CUSTOM_DATAFLOW_JOB_LOG: [{}] caught exception: {}\'.format(command, e))


REQUIRED_PACKAGES = [
]

print(\"======\\nRunning setup.py\\n==========\")
setuptools.setup(
    name=\'main_setup\',
    version=\'1.0.0\',
    description=\'DataFlow worker\',
    install_requires=REQUIRED_PACKAGES,
    packages=setuptools.find_packages(),
    cmdclass={
        \'install\': install,
        \'CustomCommands\': CustomCommands,
        }
    )```

标签： python oracle google-cloud-platform google-bigquery google-cloud-dataflow

【解决方案1】：

您是否已验证数据流工作人员肯定可以访问该 gcs 存储桶？这可能会导致这里出现问题。

一般来说，我相信这类事情的推荐路径是使用 --extra_package 标志 - https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/#local-or-nonpypi - 你可能会有更多的运气这样做。

【讨论】：