带有本地数据流服务器的多语言处理器

Question

我一直在尝试使用polyglot并构建一个简单的python处理器。我遵循了polyglot recipe，但我无法获得要部署的流。我最初部署的处理器与示例中使用的处理器相同，但得到以下错误：

请求的未知命令行参数:请求的spring.cloud.stream.bindings.input.destination未知环境变量: SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS

Traceback (most recent call last):
File "/processor/python_processor.py", line 10, in
consumer = KafkaConsumer(get_input_channel(), bootstrap_servers=[get_kafka_binder_brokers()])
File "/usr/local/lib/python2.7/dist-packages/kafka/consumer/group.py", line 353, in init
self._client = KafkaClient(metrics=self._metrics, **self.config)
File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 203, in init
self.cluster = ClusterMetadata(**self.config)
File "/usr/local/lib/python2.7/dist-packages/kafka/cluster.py", line 67, in init
self._bootstrap_brokers = self._generate_bootstrap_brokers()
File "/usr/local/lib/python2.7/dist-packages/kafka/cluster.py", line 71, in _generate_bootstrap_brokers
bootstrap_hosts = collect_hosts(self.config['bootstrap_servers'])
File "/usr/local/lib/python2.7/dist-packages/kafka/conn.py", line 1336, in collect_hosts
host, port, afi = get_ip_port_afi(host_port)
File "/usr/local/lib/python2.7/dist-packages/kafka/conn.py", line 1289, in get_ip_port_afi
host_and_port_str = host_and_port_str.strip()
AttributeError: 'NoneType' object has no attribute 'strip'
Exception AttributeError: "'KafkaClient' object has no attribute '_closed'" in <bound method KafkaClient.del of <kafka.client_async.KafkaClient object at 0x7f8b7024cf10>> ignored

然后，我尝试通过部署流传递环境和绑定参数，但这不起作用。当我手动将spring.cloud.stream.bindings.input.destination和SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS参数插入到Kafka的使用者中时，我能够部署流作为一种变通方法。我不完全确定是什么导致了这个问题，在Kubernetes上部署这个会有什么不同，或者这是Polyglot和Dataflow的问题吗？这方面的任何帮助都将不胜感激。

再现步骤:尝试在本地数据流服务器上部署polyglot配方中的polyglot处理器流。我还使用了与示例中相同的流定义: http --server.port=32123 | python-processor --reversestring=true | log。

附加上下文:我正在尝试将流部署到SPDF和Kafka的本地安装上，因为我在使用Docker部署自定义python应用程序时遇到了一些问题。

Answer 1

您在上面发布的方法需要SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS环境变量作为服务器配置的一部分(因为流是通过Skipper服务器管理的，所以您需要在Skipper服务器配置中设置此环境变量)。

您可以查看this文档，了解如何在Skipper服务器部署中将SPRING_CLOUD_STREAM_KAFKA_BINDER_BROKERS设置为环境属性。

在部署python-processor流应用程序时，也可以将此属性作为部署器属性进行传递。您可以参考this文档，了解如何在流部署时传递部署属性来设置Spring Cloud Stream属性(这里是绑定器配置属性)。