Python Apache Beam 错误 \"InvalidSchema: No connection adapters were found for\" when request api url with spaces

Question

以下来自 Apache Beam Pipeline to read from REST API runs locally but not on Dataflow pipeline 的示例从 api 请求数据

response = requests.get(url, auth=HTTPDigestAuth(self.USER, self.PASSWORD), headers=headers)

网址字符串在哪里

url = "https://host:port/car('power%203')/speed"

管道因错误而失败，请注意 'power%203 周围的额外 \：

InvalidSchema: No connection adapters were found for '(("https://host:post/car(\'power%203\')/speed",),)' [while running 'fetch API data']

想法是在本地开发和测试管道，然后在 gcp 数据流上运行生产。请求在管道外工作，但在 Python Apache Beam 管道内失败。在 DirectRunner 上从 WSL2 Ubuntu conda pyhton 3.9 环境或云 jupyter hub 执行的管道仍然返回相同的错误。请在下面找到完整的管道示例：

import logging
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, StandardOptions
import requests
import json
from requests.auth import HTTPDigestAuth

class get_api_data(beam.DoFn):
    def __init__(self, url):
        self.url = url,
        self.USER = 'user' 
        self.PASSWORD = 'password'

    def process(self, buffer=[]):        
        logging.info(self.url)
        headers = {
            'Prefer': f'data.maxpagesize=2000',
        }        
        response = requests.get(self.url, auth=HTTPDigestAuth(self.USER, self.PASSWORD), headers=headers)
        buffer = response.json()['value']
        return buffer


class Split(beam.DoFn):
    def process(self, element):
        try:
            etag = element['etag']
            car_id = element['carID']
            power = element['power']
            speed = element['speed']
        except ValueError as e:
            logging.error(e)

        return [{
            'etag': str(etag),
            'car_id': str(car_id),
            'power': int(power),
            'speed': float(speed),
        }]

def run(argv=None):   
    url = "https://host:port/car('power%203')/speed"
    p1 = beam.Pipeline(options=pipeline_options)
    ingest_data = (
        p1
        | 'Start Pipeline' >> beam.Create([None])
        | 'fetch API data' >> beam.ParDo(get_api_data(url)) 
        | 'split records' >> beam.ParDo(Split())
        | 'write to text' >> beam.io.WriteToText("./test_v2.csv")
    )

    result = p1.run()

if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()

这让我真的很困惑，如果有人可以分享关于为什么 url 字符串被扭曲的任何建议或 cmets，我将不胜感激。

Answer 1

删除 get_api_data 类中 url 旁边的逗号 - 它应该可以解决问题

class get_api_data(beam.DoFn):
    def __init__(self, url):
        self.url = url
        self.USER = 'user' 
        self.PASSWORD = 'password'