zoukankan      html  css  js  c++  java
  • kafka--通过python操作topic

    修改 topic 的分区数

    shiyanlou:bin/ $ ./kafka-topics.sh --zookeeper localhost:2181 --alter --topic mySendTopic --partitions 4
    WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
    Adding partitions succeeded!

    如果 topic 的 partition 数量增加了,那么消息的分区逻辑或者消息的顺序都会受到影响。这一点我们在学习 consumer 课程中详解。

    查看是否增加了:

    shiyanlou:bin/ $ ./kafka-topics.sh --zookeeper localhost:2181 --describe --topic mySendTopic           [15:00:54]
    Topic:mySendTopic PartitionCount:4 ReplicationFactor:2 Configs:
     Topic: mySendTopic Partition: 0 Leader: 2 Replicas: 2,0 Isr: 2,0
     Topic: mySendTopic Partition: 1 Leader: 0 Replicas: 0,1 Isr: 0,1
     Topic: mySendTopic Partition: 2 Leader: 1 Replicas: 1,2 Isr: 1,2
     Topic: mySendTopic Partition: 3 Leader: 2 Replicas: 2,1 Isr: 2,1

    对于一个topic来说,其partition的数量,只能增加,不能减少

    删除topic:

    shiyanlou:bin/ $./kafka-topics.sh --zookeeper localhost:2181 --delete --topic myDeleteTopic

    myDeleteTopic 只是标记为删除,没有真正删除。要想彻底删除,需在 server.properties 中设置 delete.topic.enable=true。运行如下命令验证这一点:./kafka-topics.sh --zookeeper localhost:2181 --list

    安装python操作kafka的包

    shiyanlou:bin/ $ sudo pip install pykafka

    shiyanlou:~/ $ cat pykafk.py                                         [16:10:15]
    #!/usr/bin/python
    
    from pykafka import KafkaClient
    client = KafkaClient(hosts="127.0.0.1:9092,127.0.0.1:9093, 127.0.0.1:9094")
    #查看主题
    print client.topics
    #查看brokers
    print client.brokers topic = client.topics['mySendTopic'] for n in client.brokers: host = client.brokers[n].host port = client.brokers[n].port id = client.brokers[n].port print "host=%s |port=%s|broker.id=%s" %(host,port,id) shiyanlou:~/ $ python pykafk.py [16:10:31] {'myFirstTopic': None, 'mySendTopic': None} {0: <pykafka.broker.Broker at 0x7ff88c363590 (host=857dafc13648, port=9092, id=0)>, 1: <pykafka.broker.Broker at 0x7ff88c363dd0 (host=857dafc13648, port=9093, id=1)>, 2: <pykafka.broker.Broker at 0x7ff88c376110 (host=857dafc13648, port=9094, id=2)>} host=857dafc13648 |port=9092|broker.id=9092 host=857dafc13648 |port=9093|broker.id=9093 host=857dafc13648 |port=9094|broker.id=9094
  • 相关阅读:
    461. Hamming Distance
    Myeclipse中WebServlet cannot be resolved to a type报错
    注解方式配置Servlet(Servlet3.0)
    Oracle连接池操作
    最短路径算法
    编写学生类Stu
    编写程序,统计某旅馆住宿客人的总数,要求输入客人姓名,输出客人编号(按先后顺序自动生成),姓名以及总人数。
    货物管理系统
    c# 利用动态库DllImport("kernel32")读写ini文件(提供Dmo下载)
    shut
  • 原文地址:https://www.cnblogs.com/kingleoric/p/8027869.html
Copyright © 2011-2022 走看看