KSQL Cli 工作，但 KSQL Python API 在 KSQL 服务器运行时不工作

Question

我已设法设置 KSQL 服务器和 KSQL CLI 客户端，以使用 KSQL 消费流并使用 cli 工作。例如显示主题命令在下面工作

ksql> SHOW TOPICS;

 Kafka Topic                            | Registered | Partitions | Partition Replicas | Consumers | Consumer Groups 
---------------------------------------------------------------------------------------------------------------------
 _confluent-ksql-default__command_topic | true       | 1          | 1                  | 0         | 0              

但是当我使用ksql python 包调用相同的SHOW TOPIC 或任何SELECT 时，查询下面给出的错误。

>>> client = KSQLAPI('http://localhost:8088')
>>> results = client.query('SHOW TOPICS;')
>>> for item in results:
...     print(item)
... 
Traceback (most recent call last):
  File "/home/ubuntu/venv_language/lib/python3.6/site-packages/ksql/api.py", line 170, in _request
    r = urllib.request.urlopen(req, timeout=self.timeout)
  File "/usr/lib/python3.6/urllib/request.py", line 223, in urlopen
    return opener.open(url, data, timeout)
  File "/usr/lib/python3.6/urllib/request.py", line 532, in open
    response = meth(req, response)
  File "/usr/lib/python3.6/urllib/request.py", line 642, in http_response
    'http', request, response, code, msg, hdrs)
  File "/usr/lib/python3.6/urllib/request.py", line 570, in error
    return self._call_chain(*args)
  File "/usr/lib/python3.6/urllib/request.py", line 504, in _call_chain
    result = func(*args)
  File "/usr/lib/python3.6/urllib/request.py", line 650, in http_error_default
    raise HTTPError(req.full_url, code, msg, hdrs, fp)
urllib.error.HTTPError: HTTP Error 400: Bad Request

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/ubuntu/venv_language/lib/python3.6/site-packages/ksql/client.py", line 63, in query
    yield from process_query_result(results, return_objects)
  File "/home/ubuntu/venv_language/lib/python3.6/site-packages/ksql/utils.py", line 100, in process_query_result
    yield from results
  File "/home/ubuntu/venv_language/lib/python3.6/site-packages/ksql/api.py", line 113, in query
    endpoint="query", sql_string=query_string, stream_properties=stream_properties
  File "/home/ubuntu/venv_language/lib/python3.6/site-packages/ksql/api.py", line 178, in _request
    raise KSQLError(content.get("message"), content.get("error_code"), content.get("stackTrace"))
ksql.errors.KSQLError: ('HTTP 415 Unsupported Media Type', None, ['org.glassfish.jersey.server.internal.routing.MethodSelectingRouter.getMethodRouter(MethodSelectingRouter.java:478)', 'org.glassfish.jersey.server.internal.routing.MethodSelectingRouter.access$000(MethodSelectingRouter.java:94)', 'org.glassfish.jersey.server.internal.routing.MethodSelectingRouter$4.apply(MethodSelectingRouter.java:779)', 'org.glassfish.jersey.server.internal.routing.MethodSelectingRouter.apply(MethodSelectingRouter.java:371)', 'org.glassfish.jersey

这是 KSQL 服务器信息

{ KsqlServerInfo: { version: "0.13.0", kafkaClusterId: "NXj5kHsORuCShpGsZWMvKg", ksqlServiceId: "confluent_rmoff_01", serverStatus: "RUNNING" } }

我在互联网上找不到与此相关的太多帮助。如果我做错了什么，有人可以在这里帮助我吗？

谢谢

Answer 1

对于SHOW TOPICS，您需要使用ksql 方法，而不是query 方法：

╰─$ python
Python 3.7.6 (default, Jan  8 2020, 11:09:02)
[Clang 11.0.0 (clang-1100.0.33.16)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> from ksql import KSQLAPI
>>> client = KSQLAPI('http://localhost:8088')
>>> client.ksql('show topics')
[{'@type': 'kafka_topics', 'statementText': 'show topics;', 'topics': [{'name': 'movements', 'replicaInfo': [1]}], 'warnings': []}]

如果你使用query 方法，你会看到你遇到的错误：

>>> results = client.query('SHOW TOPICS;')
>>> for item in results:
...     print(item)
...
Traceback (most recent call last):
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/site-packages/ksql/api.py", line 170, in _request
    r = urllib.request.urlopen(req, timeout=self.timeout)
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/urllib/request.py", line 222, in urlopen
    return opener.open(url, data, timeout)
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/urllib/request.py", line 531, in open
    response = meth(req, response)
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/urllib/request.py", line 641, in http_response
    'http', request, response, code, msg, hdrs)
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/urllib/request.py", line 569, in error
    return self._call_chain(*args)
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/urllib/request.py", line 503, in _call_chain
    result = func(*args)
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/urllib/request.py", line 649, in http_error_default
    raise HTTPError(req.full_url, code, msg, hdrs, fp)
urllib.error.HTTPError: HTTP Error 400: Bad Request

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/site-packages/ksql/client.py", line 63, in query
    yield from process_query_result(results, return_objects)
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/site-packages/ksql/utils.py", line 100, in process_query_result
    yield from results
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/site-packages/ksql/api.py", line 113, in query
    endpoint="query", sql_string=query_string, stream_properties=stream_properties
  File "/Users/rmoff/.pyenv/versions/3.7.6/lib/python3.7/site-packages/ksql/api.py", line 178, in _request
    raise KSQLError(content.get("message"), content.get("error_code"), content.get("stackTrace"))
ksql.errors.KSQLError: ("Statement type `io.confluent.ksql.parser.KsqlParser$PreparedStatement' not supported for this resource", 40000, None)

Answer 2

使用它可以解决你的问题：

client = KSQLAPI('http://URSERVER:8088')
client.ksql('show topics')