0

我正在尝试从 GCP 存储桶下载加密文件和密钥，然后解密文件并将其加载回存储桶。所以我构建了这个 DataFlow 管道，如下所示：

class downloadFile(beam.DoFn):
def __init__(self):
    self.bucket_name = 'bucket_name'
    self.source_blob_name = 'test.csv.gpg'
    self.destination_file_name = "/tmp/test.csv.gpg"

def process(self, element):
    from google.cloud import storage

    storage_client = storage.Client()
    bucket = storage_client.bucket(self.bucket_name)
    blob = bucket.blob(self.source_blob_name)
    blob.download_to_filename(self.destination_file_name)

这里我使用self.destination_file_name = "/tmp/test.csv.gpg"是因为我从其他人那里了解到 DataFlow 作业将在 Linux VM 上运行，因此将文件下载到这个 /tmp/ 路径是完全安全的。

class downloadKey(beam.DoFn):
def __init__(self):
    self.bucket_name = 'bucket_name'
    self.source_blob_name = 'privateKey.txt'
    self.destination_file_name = "/tmp/privateKey.txt"


def process(self, element):
    from google.cloud import storage

    storage_client = storage.Client()
    bucket = storage_client.bucket(self.bucket_name)
    blob = bucket.blob(self.source_blob_name)
    blob.download_to_filename(self.destination_file_name)

基本上，两个下载DoFns具有相同的结构。下载文件和密钥后，密钥将被导入到运行 VM 的 DataFlow：

class importKey(beam.DoFn):
def process(self, element):
    import subprocess
    subprocess.call(['gpg', '--import','/tmp/privateKey.txt'])

然后解密DoFn：

class decryption(beam.DoFn):
def process(self, element, *args, **kwargs):
    import subprocess
    subprocess.call(['gpg', '-d', '/tmp/test.csv.gpg > test.csv'])
    
    # load file back to bucket
    bucket_name = 'bucket_name'
    source_file_name = '/tmp/test.csv'
    destination_blob_name = "clearText.csv"

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)
    blob.upload_from_filename(source_file_name)

所以这个解密 DoFn 将调用带有子进程的 gpg 命令来解密文件。

最后是管道本身：

dummyMessage = {"projectID":"fakeProjectID",
            "bucketID":"fakeBucketID"}

setp= (
    p
    | 'Create Sample'
            >> beam.Create([dummyMessage["projectID"]])
    |"testDecrypt" >> beam.ParDo(downloadLookupFile())
    |"testDecrypt2" >> beam.ParDo(downloadKey())
    |"testDecrypt3" >> beam.ParDo(importKey())
    |"testDecrypt4" >> beam.ParDo(decryption())
   )

这里我只是创建一个虚拟消息来调用管道，稍后将替换为真实消息。

当我运行管道时，一切正常，我可以看到 DataFlow 中已创建作业，并且显示作业状态为成功。但在存储桶中我看不到解密的文件。

我在要调试的代码中添加了几个打印语句，似乎在 downloadFile() 和 downloadKey() 方法中，从未达到 process()，这意味着没有处理过任何文件。任何人都可以分享一些有关如何在 DoFn 中访问 GCS 存储桶的知识吗？我不确定代码的哪一部分是磨损的，对我来说一切都很好。

任何帮助将不胜感激。

4

1 回答 1

0

欢迎 Alex 使用 stackoverflow。

  • 首先，关于日志（打印语句），如果您没有看到它们，可能是因为您看错了地方。事实上，如果你把它们放在processDoFns 里面（比如decryption类），你需要看里面WORKER LOGS而不是JOB LOGS你的终端。在下面的屏幕截图中，我展示了如何访问工作日志。作业日志或驱动程序日志是显示您在管道创建（beam.Create...）级别添加的打印/日志的日志，如果您从终端运行作业，您可以在终端中看到它们。

  • 那么，恕我直言，数据流不是这种需求的正确处理平台。它用于并行分布式处理大文件块或大数据（比如说 > 2GB）。这意味着在幕后你有一个部分在一个工作节点（幕后的 GCE VM 实例）上处理，而另一块文件在另一个工作节点上处理。在您的情况下，如果您有超过 1 个工作人员，您可能会在一个节点中下载加密文件，在另一个节点中下载密钥，在第三个节点中进行解密。所以使用/tmp将是过时的。

  • 最后一个解决方案是使用例如云函数（CF），它将以单线程方式运行，并允许您重用不同process方法中的代码：

  1. CF 是通过将加密文件上传到 GCS 存储桶来触发的。这里有一些文档如何设置这样的触发器：https ://cloud.google.com/functions/docs/calling/storage （python中有示例）
  2. 您的 CF 代码将下载加密密钥、解密并将解密的文件上传回另一个 GCS 存储桶。对于云功能，您可以将内存设置为高达 8GB 并使用/tmp使用幕后内存的内存。

顺便说一句，安全方面，我认为将加密密钥存储在 GCS 中不是一个好习惯，请尝试查看https://cloud.google.com/secret-manager 在此处输入图像描述

于 2021-08-21T11:31:50.087 回答