Python操作Kafka爬坑

组内做大数据，需要kafka写入数据，最近在看python正好，练练手，网上找了一圈，都是用的pykafka，经过一整圈的安装，最终搞定，代码如下
#coding:u8
import sys
import time
import random
import datetime
import MySQLdb
import codecs
from pykafka import KafkaClient
import logging
import json
import threading
'''
******************
'''
ad=[]
try:
ini=file("set.txt")
ad=ini.readline().splitlines()
ini.close
except Exception as e:
print "open settings file Error:",type(e)
ad=["192.168.1.121:9092"]
print "open ini file"
try:
client = KafkaClient(hosts = ad[0])
print "Topics:",client.topics
topic = client.topics["mytopic"]
except Exception as e:
print "Opening kafka Error:%s" %(type(e))
sys.exit(1)
print "before threading"

try:
with tp.get_sync_producer() as producer:
producer.produce(str(dct2))
except Exception as e:
print "Error:" ,type(e)

print "ini consumer"
while 1==1:
print "nn",type(consumer)
for message in consumer:
print "mm"
if message is not None:
print message.offset, message.value
except Exception as e:
print e,type(e)

运行结果，可以列出topic，写入的数据也没有报错信息。但是，消费者取不到数据，无论是kafka直接取，还是python写消费者代码。
后来采用了 kafkapython 正常，代码如下：

#coding:utf-8
import sys
import time
import random
import datetime
import codecs
import kafka.kafkaProducer
import logging
import json
import threading

ad=[]
try:
ini=file("set.txt")
ad=ini.readline().splitlines()
ini.close
except Exception as e:
ad=["192.168.1.121:9092,192.168.1.122.9092"]
#print "open settings file Error:%d,%s" %(e.args[0],e.args[1])
print "Opening settings file Error:",e,type(e)
print "Opened ini file"
'''
try:
client = KafkaClient(hosts = ad[0])
print "Topics:",client.topics
topic = client.topics["mytopic"]
except Exception as e:
print "Opening kafka Error:%s" %(e.args[0])
sys.exit(1)
print "before threading"
'''
try:
producer = KafkaProducer(bootstrap_servers=ad[0], value_serializer=lambda m: json.dumps(m).encode('utf-8'))
except Exception as e:
print "Opening kafka Error:",e,type(e)
sys.exit(1)

print "before threading"

threads=[]
for i in range(0,12):
try:
threads.append(threading.Thread(target=tf,args=(producer,i)))
threads[i].start()
except Exception as e:
print "Treand error at Thread:%d:%s,%s" %(i,e,type(e))

print "main thread is ended"

代码均有所节略。

Python操作Kafka爬坑相关推荐

Python 操作 Kafka --- kafka-python
kafka-python:https://github.com/dpkp/kafka-python kafka-python 文档:https://kafka-python.readthedocs.i ...
kafka实战教程(python操作kafka)，kafka配置文件详解
全栈工程师开发手册 (作者:栾鹏) 架构系列文章应用往Kafka写数据的原因有很多:用户行为分析.日志存储.异步通信等.多样化的使用场景带来了多样化的需求:消息是否能丢失?是否容忍重复?消息的吞吐量 ...
Python操作kafka实现数据定时推送和获取，并解决中文乱码问题
上篇介绍了kafka环境的搭建以及实现了Python如何简单操作kafka.本篇实现如何定时向kafka推送和接收数据,并解决了中文乱码问题. 设置Producer每隔5秒推送一次数据,Consume ...
python操作mysql数据库—坑吭
一.原始代码 #coding=utf-8import MySQLdb from cProfile import Profileclass MySQLUtils():def __init__(self) ...
python使用kafka原理详解_Python操作Kafka原理及使用详解
Python操作Kafka原理及使用详解一.什么是Kafka Kafka是一个分布式流处理系统,流处理系统使它可以像消息队列一样publish或者subscribe消息,分布式提供了容错性,并发处理 ...
使用Python读写kafka
目录 1. Kafka安装与使用 1.1 下载 1.2 安装 1.3 配置 1.4 运行 1.4.1 启动zookeeper 1.4.2 启动kafka 1.5 第一个消息 1.5.1 创建一个top ...
使用python读取kafka实时topic数据demo，包括安装kafka module
1. 安装kafka module kafka-python安装,转载:https://blog.csdn.net/see_you_see_me/article/details/78468421 1. ...
python操作excel及爬取美赛证书
今天美赛成绩出来了,S奖,有点遗憾.但工作还是要继续,因为要下载校队的获奖证书,所以使用python爬虫来爬取证书,省点事. python操作excel 在这里我选择pandas库来操作excel,详 ...
python闲鱼爬虫_Python 爬虫爬坑路（二）——B站图片，咸鱼的正确 GET 姿势
前言昨天在写完入门级爬虫之后 ,马上就迫不及待的着手开始写 B站的图片爬虫了,真的很喜欢这个破站呢 (〜￣△￣)〜这里不涉及到 Python 爬虫的高级技巧,没有使用框架,没有考虑反爬机制,没有 ...

Python操作Kafka爬坑

Python操作Kafka爬坑相关推荐

最新文章

热门文章