我是Kafka的新手并在论坛中搜索了不同的帖子,但找不到解决方案 . 我已经在EC2实例上安装了kafka,并尝试从我的ubuntu本地机器上连接它 . 我的目标是让python kafka客户端(包括Producer和Consumer)在我的本地机器上运行,并通过EC2 kafka实例发送/接收数据 . 那可能吗?
Properties set in server.properties config file:
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://<ec2-public-DNS>:9092
On Kafka EC2 Instance:
netstat -an | grep LISTEN
tcp 0 0 0.0.0.0:22 0.0.0.0:* LISTEN
tcp6 0 0 :::9092 :::* LISTEN
On Zookeeper cli on Kafka EC2 Instance:
get /brokers/ids/0
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://<ec2-public-DNS>:9092"],"jmx_port":-1,"host":"<ec2-public-DNS>","timestamp":"1492900361516","port":9092,"version":4}
cZxid = 0xed
ctime = Sat Apr 22 22:32:41 UTC 2017
mZxid = 0xed
mtime = Sat Apr 22 22:32:41 UTC 2017
pZxid = 0xed
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x15b97cb9d060000
dataLength = 250
numChildren = 0
Python client(Producer) on my local machine:
from kafka import KafkaProducer
import time
import json
producer = KafkaProducer(bootstrap_servers="<ec2-public-DNS>:9092")
for i in range(100):
dict = {}
dict['name_'+str(i)] = 'FILE_' + str(i)
dict['size_'+str(i)] = '23.' + str(i)
dict['host_'+str(i)] = '10.0.0.0' + str(i)
jd = json.dumps(dict)
producer.send('console-test-topic', jd)
time.sleep(2)
Python client(Consumer) on my local machine:
from kafka import KafkaConsumer
consumer = KafkaConsumer('console-test-topic', bootstrap_servers="<ec2-public-DNS>:9092")
for msg in consumer:
print (msg)
但是, 生产环境 者无法连接到Kafka EC2实例并且失败并出现以下错误:
**kafka.errors.NoBrokersAvailable: NoBrokersAvailable**
Please refer to the link for my security group rules:
goo.gl/ZUVknv
Running Producer in debug mode on my local machine:
DEBUG:kafka.producer.kafka:Starting the Kafka producer
DEBUG:kafka.metrics.metrics:Added sensor with name connections-closed
DEBUG:kafka.metrics.metrics:Added sensor with name connections-created
DEBUG:kafka.metrics.metrics:Added sensor with name select-time
DEBUG:kafka.metrics.metrics:Added sensor with name io-time
INFO:kafka.client:Bootstrapping cluster metadata from [('ec2-54-91-87-14.compute-1.amazonaws.com', 9092, 0)]
DEBUG:kafka.client:Attempting to bootstrap via node at ec2-54-91-87-14.compute-1.amazonaws.com:9092
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent-received
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name request-latency
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-bootstrap.latency
DEBUG:kafka.client:Node bootstrap connected
DEBUG:kafka.cluster:Updated cluster metadata to ClusterMetadata(brokers: 1, topics: 2, groups: 0)
INFO:kafka.client:Bootstrap succeeded: found 1 brokers and 2 topics.
DEBUG:kafka.client:Initiating connection to node 0 at ec2-54-91-87-14.compute-1.amazonaws.com:9092
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.bytes-sent
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.bytes-received
DEBUG:kafka.metrics.metrics:Added sensor with name node-0.latency
INFO:kafka.producer.kafka:Kafka producer closed
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/usr/local/lib/python2.7/dist-packages/kafka/producer/kafka.py", line 335, in __init__
**self.config)
File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 210, in __init__
self.config['api_version'] = self.check_version(timeout=check_timeout)
File "/usr/local/lib/python2.7/dist-packages/kafka/client_async.py", line 828, in check_version
raise Errors.NoBrokersAvailable()
kafka.errors.NoBrokersAvailable: NoBrokersAvailable
我尝试在另一个EC2实例中运行 生产环境 者客户端(在与kafka实例相同的VPN中)并且它工作正常 . 但是,当 生产环境 者在我的本地机器上运行时,它不起作用 . 'advertised.listeners'属性是否在同一个(AWS VPN)网络中宣传kafka经纪人?或者我也可以从本地机器连接它吗?如果有人能指出我正确的方向,请告诉我 .
2 回答
几个月前我经历了类似的事情,基本上我在N. Virginia有一个ec2实例和Kafka,我在本地机器上配置了topbeat,以便将指标发送到该ec2实例 . 我能够通过添加来实现它
作为kafka的server.properties中的配置,但根据documentation,不推荐使用此属性 .
进一步阅读documentation,据说如果您处于IaaS环境中,则必须配置与代理绑定的接口不同的advertised.listeners .
您是否有机会忽略将AWS安全组中的入站端口
9092
打开到本地网络的外部IP?如果您选择向所有人开放,请将其打开至0.0.0.0/0
(但请注意安全隐含) .