我可以从 Python 调用 Bluemix 消息中心服务吗? [英] Can I call the Bluemix message hub service from Python?

问题描述

kafka-python 客户端支持 Kafka 0.9 但不支持t 显然包括新的身份验证和加密功能，所以我的猜测是它只适用于开放服务器(如在以前的版本中).在任何情况下，即使是 Java 客户端也需要一个特殊的消息中心登录模块来连接(或者从示例中看起来是这样)这表明除非有可用于 Python 的类似模块，否则什么都不会工作.

The kafka-python client supports Kafka 0.9 but doesn't obviously include the new authentication and encryption features so my guess is that it only works with open servers (as in previous releases). In any case, even the Java client needs a special message hub login module to connect (or so it would seem from the example) which suggests that nothing will work unless there is a similar module available for Python.

我的具体场景是，我想使用同样托管在 Bluemix(Apache Spark 服务)中的 Jupyter 笔记本中的消息中心服务.

My specific scenario is that I want to use the message hub service from a Jupyter notebook also hosted in Bluemix (the Apache Spark service).

推荐答案

我能够使用 kafka-python 库进行连接:

I was able to connect using the kafka-python library:

$ pip install --user kafka-python

然后……

from kafka import KafkaProducer
from kafka.errors import KafkaError
import ssl

############################################
# Service credentials from Bluemix UI:
############################################
bootstrap_servers =   # kafka_brokers_sasl
sasl_plain_username = # user
sasl_plain_password = # password
############################################

sasl_mechanism = 'PLAIN'
security_protocol = 'SASL_SSL'

# Create a new context using system defaults, disable all but TLS1.2
context = ssl.create_default_context()
context.options &= ssl.OP_NO_TLSv1
context.options &= ssl.OP_NO_TLSv1_1

producer = KafkaProducer(bootstrap_servers = bootstrap_servers,
                         sasl_plain_username = sasl_plain_username,
                         sasl_plain_password = sasl_plain_password,
                         security_protocol = security_protocol,
                         ssl_context = context,
                         sasl_mechanism = sasl_mechanism,
                         api_version=(0,10))

# Asynchronous by default
future = producer.send('my-topic', b'raw_bytes')

# Block for 'synchronous' sends
try:
    record_metadata = future.get(timeout=10)
except KafkaError:
    # Decide what to do if produce request failed...
    log.exception()
    pass

# Successful result returns assigned partition and offset
print (record_metadata.topic)
print (record_metadata.partition)
print (record_metadata.offset)

这对我从 Bluemix spark as a service from a jupyter notebook 起作用，但是请注意，这种方法没有使用 spark.代码只是在驱动程序主机上运行.

This worked for me from Bluemix spark as a service from a jupyter notebook, however, note that this approach is not using spark. The code is just running on the driver host.

这篇关于我可以从 Python 调用 Bluemix 消息中心服务吗?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！