我想使用带有 python beautifulsoup 的 kafka 生产者向 kafka 代理发送消息答案

【问题标题】：I want to use kafka producer with python beautifulsoup to send message to kafka broker我想使用带有 python beautifulsoup 的 kafka 生产者向 kafka 代理发送消息
【发布时间】：2020-08-23 12:34:30
【问题描述】：

我正在使用kafka-python和BeautifulSoup来抓取我经常进入的网站，并向带有python生产者的kafka broker发送消息。

我想做的是每当新帖子上传到网站上（实际上它是某种像reddit这样的社区，通常是韩国嘻哈粉丝用来分享信息等），该帖子应该发送给kafka经纪人。

但是，我的问题在 while 循环内，只有最新的帖子不断重复发送到 kafka 代理。这不是我想要的。

另外，第二个问题是加载新帖子时，

HTTP 错误 502：发生错误网关错误

soup = BeautifulSoup(urllib.request.urlopen("http://hiphople.com/kboard").read(), "html.parser")

消息不再发送。

这是dataScraping.py

from bs4 import BeautifulSoup
import re
import urllib.request

pattern = re.compile('[0-9]+')

def parseContent():
    soup = BeautifulSoup(urllib.request.urlopen("http://hiphople.com/kboard").read(), "html.parser")
    for div in soup.find_all("tr", class_="notice"):
        div.decompose()

    key_num = pattern.findall(soup.find_all("td", class_="no")[0].text)
    category = soup.find_all("td", class_="categoryTD")[0].find("span").text
    author = soup.find_all("td", class_="author")[0].find("span").text
    title = soup.find_all("td", class_="title")[0].find("a").text
    link = "http://hiphople.com" + soup.find_all("td", class_="title")[0].find("a").attrs["href"]

    soup2 = BeautifulSoup(urllib.request.urlopen(link).read(), "html.parser")
    content = str(soup2.find_all("div", class_="article-content")[0].find_all("p"))
    content = re.sub("<.+?>","", content,0).strip()
    content = re.sub("\xa0","", content, 0).strip()

    result = {"key_num": key_num, "catetory": category, "title": title, "author": author, "content": content}
    return result

if __name__ == "__main__":
    print("data scraping from website")

这是 PythonWebScraping.py

import json
from kafka import KafkaProducer
from dataScraping import parseContent

def json_serializer(data):
    return json.dumps(data).encode("utf-8")


producer = KafkaProducer(acks=1, compression_type = "gzip", bootstrap_servers=["localhost:9092"],
                         value_serializer = json_serializer)
    
if __name__ == "__main__":
    while (True):
        result = parseContent()
        producer.send("hiphople",result)

请让我知道如何修复我的代码，以便我可以按预期将新创建的帖子发送到 kafka 代理。

【问题讨论】：

标签： python beautifulsoup kafka-python

【解决方案1】：

您的功能正在运行，但您确实只返回一个事件，我没有得到 502 bad gateway，也许您因为访问 url 的次数过多而将其作为 ddos 保护，尝试添加 delays/sleep ，或者您的ip 已被禁止以阻止它抓取 url...

对于您的第二个错误，您的函数仅返回一条/最后一条消息

您每次都将结果发送给 kafka，这就是为什么您一遍又一遍地看到相同的消息，

你正在抓取最后一个事件，你希望你的函数做什么？

prevResult = ""
而（真）：
结果 = parseContent()
如果（上一个结果！=结果）：
prevResult = 结果
打印（结果）

【讨论】：