如何在 Apache Beam 中解析 2 个 json 文件

Question

我有 2 个 json 配置文件要读取并希望将值分配给变量。我正在使用 apache Beam 创建数据流作业，但无法解析这些文件并将值分配给变量。

config1.json - { "bucket_name": "mybucket"} config2.json - { "dataset_name": "mydataset"}

这是管道语句 ---- 我首先尝试使用一个 JSON 文件，但即使这样也不起作用

with beam.Pipeline(options=pipeline_options) as pipeline:
        steps = (pipeline
                | "Getdata" >> beam.io.ReadFromText(custom_options.configfile)
                | "CUSTOM JSON PARSE" >> beam.ParDo(custom_json_parser(custom_options.configfile))
                | "write to GCS" >> beam.io.WriteToText('gs://mynewbucket/outputfile.txt')
            )
    result = pipeline.run()
    result.wait_until_finish()

我还尝试创建一个函数来解析至少一个文件。这是我创建的示例方法，但它不起作用。

class custom_json_parser(beam.DoFn):
    import apache_beam as beam
    from apache_beam.io.gcp import gcsio
    import logging
    def __init__(self, configfile):
        self.configfile = configfile
    def process(self, configfile):
        logging.info("JSON PARSING STARTED")
        with beam.io.gcp.gcsio.GcsIO().open(self.configfile, 'r') as f:
            for line in f:
                data = json.loads(line)
                bucket = data.get('bucket_name')
        dataset = data.get('dataset_name') ```
        
  Can someone please suggest the best method to resolve this issue in apache beam?

Thanks in Advance          

Answer 1

如果您只需要在管道中读取一次文件，请不要在管道中读取它们，而是在运行之前读取它们。

• 从 GCS 读取文件
• 解析文件并将有用的内容放入管道选项映射中
• 运行您的管道并使用选项中的数据

---

编辑 1

您可以在管道之前使用这段代码加载文件并读取它。简单的 Python，标准 GCS 库。

    from google.cloud import storage
    import json

    client = storage.Client()
    bucket = client.get_bucket('your-bucket')
    blob = bucket.get_blob("name.json")

    json_data = blob.download_as_string().decode('UTF-8')
    print(json_data) # print -> {"name": "works!!"}
    print(json.loads(json_data)["name"]) # print -> works!!

Answer 2

你可以试试下面的代码sn-p：-

解析文件的功能

class custom_json_parser(beam.DoFn):
    def process(self, element):
        logging.info(element)
        data = json.loads(element)        
        bucket = data.get('bucket_name')
        dataset = data.get('dataset_name')
        return [{"bucket": bucket , "dataset": dataset }]

通过管道你可以调用函数

with beam.Pipeline(options=pipeline_options) as pipeline:
        steps = (pipeline
                | "Getdata" >> beam.io.ReadFromText(custom_options.configfile)
                | "CUSTOM JSON PARSE" >> beam.ParDo(custom_json_parser())
                | "write to GCS" >> beam.io.WriteToText('gs://mynewbucket/outputfile.txt')
            )
    result = pipeline.run()
    result.wait_until_finish()

它会起作用的。