ImportError：尝试将数据提交到数据流中的 firestore 时，没有名为 firebase_admin 的模块

Question

我目前正在尝试将数据从我的 python apache beam 数据流管道提交到 firestore。

但我一直收到错误：

Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/batchworker.py", line 609, in do_work
    work_executor.execute()
  File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/executor.py", line 167, in execute
    op.start()
  File "dataflow_worker/shuffle_operations.py", line 49, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start
    def start(self):
  File "dataflow_worker/shuffle_operations.py", line 50, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start
    with self.scoped_start_state:
  File "dataflow_worker/shuffle_operations.py", line 65, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start
    with self.shuffle_source.reader() as reader:
  File "dataflow_worker/shuffle_operations.py", line 69, in dataflow_worker.shuffle_operations.GroupedShuffleReadOperation.start
    self.output(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 159, in apache_beam.runners.worker.operations.Operation.output
    cython.cast(Receiver, self.receivers[output_index]).receive(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 85, in apache_beam.runners.worker.operations.ConsumerSet.receive
    cython.cast(Operation, consumer).process(windowed_value)
  File "dataflow_worker/shuffle_operations.py", line 233, in dataflow_worker.shuffle_operations.BatchGroupAlsoByWindowsOperation.process
    self.output(wvalue.with_value((k, wvalue.value)))
  File "apache_beam/runners/worker/operations.py", line 159, in apache_beam.runners.worker.operations.Operation.output
    cython.cast(Receiver, self.receivers[output_index]).receive(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 85, in apache_beam.runners.worker.operations.ConsumerSet.receive
    cython.cast(Operation, consumer).process(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 392, in apache_beam.runners.worker.operations.DoOperation.process
    with self.scoped_process_state:
  File "apache_beam/runners/worker/operations.py", line 393, in apache_beam.runners.worker.operations.DoOperation.process
    self.dofn_receiver.receive(o)
  File "apache_beam/runners/common.py", line 488, in apache_beam.runners.common.DoFnRunner.receive
    self.process(windowed_value)
  File "apache_beam/runners/common.py", line 496, in apache_beam.runners.common.DoFnRunner.process
    self._reraise_augmented(exn)
  File "apache_beam/runners/common.py", line 521, in apache_beam.runners.common.DoFnRunner._reraise_augmented
    raise
  File "apache_beam/runners/common.py", line 494, in apache_beam.runners.common.DoFnRunner.process
    self.do_fn_invoker.invoke_process(windowed_value)
  File "apache_beam/runners/common.py", line 395, in apache_beam.runners.common.PerWindowInvoker.invoke_process
    self._invoke_per_window(
  File "apache_beam/runners/common.py", line 432, in apache_beam.runners.common.PerWindowInvoker._invoke_per_window
    output_processor.process_outputs(
  File "apache_beam/runners/common.py", line 561, in apache_beam.runners.common._OutputProcessor.process_outputs
    def process_outputs(self, windowed_input_element, results):
  File "apache_beam/runners/common.py", line 592, in apache_beam.runners.common._OutputProcessor.process_outputs
    self.main_receivers.receive(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 85, in apache_beam.runners.worker.operations.ConsumerSet.receive
    cython.cast(Operation, consumer).process(windowed_value)
  File "apache_beam/runners/worker/operations.py", line 392, in apache_beam.runners.worker.operations.DoOperation.process
    with self.scoped_process_state:
  File "apache_beam/runners/worker/operations.py", line 393, in apache_beam.runners.worker.operations.DoOperation.process
    self.dofn_receiver.receive(o)
  File "apache_beam/runners/common.py", line 488, in apache_beam.runners.common.DoFnRunner.receive
    self.process(windowed_value)
  File "apache_beam/runners/common.py", line 496, in apache_beam.runners.common.DoFnRunner.process
    self._reraise_augmented(exn)
  File "apache_beam/runners/common.py", line 537, in apache_beam.runners.common.DoFnRunner._reraise_augmented
    six.raise_from(new_exn, original_traceback)
  File "/usr/local/lib/python2.7/dist-packages/six.py", line 718, in raise_from
    raise value
ImportError: No module named firebase_admin [while running 'Batches to Firestore']

我目前正在函数内部调用 firebase_admin ，尽管这与在文件开头调用它相比并没有什么区别。

class FireBatch(beam.DoFn):
    def process(self, element):
        """
        Make a batch element and send to firestore
        """
        import firebase_admin
        from firebase_admin import credentials
        from firebase_admin import firestore

        cred = credentials.Certificate('./creds/pipeCreds.json')
        firebase_admin.initialize_app(cred)

        db = firestore.Client()

有没有办法让数据流识别管道内的 firebase_admin ？或者将数据从管道提交到firestore？

我已经安装了所有必要的软件包，升级了 pip，刷新了凭据，尝试了 google.cloud.firestore 而不是 fire_base admin。

感谢您的帮助！

Answer 1

这个问题是由于没有使用 Google SDK 进行初始化，我尝试使用 GitBash 进行初始化，显然配置设置不正确。

下一步是使用requirements.txt 文件初始化应用程序。

pip freeze > requirements.txt

之后，仅在requirements.txt文件中包含应用程序所需的文件，并通过添加以下内容来初始化应用程序：

--requirements_file requirements.txt

另外，有必要将其更改为 ，firebase_admin==2.11.0因为firebase_admin==2.8.0这会引发错误。