python开发，spark接收kafka的输入流时

Z时代
2024-02-24
分类：IT

import cv2
import sys
import findspark
findspark.init()
from kafka import KafkaConsumer
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils,TopicAndPartition
if __name__ =="__main__":
    print('kaishi')
    sc = SparkContext(appName="videostreamprocessor")
    ssc = StreamingContext(sc,1)
    brokers="localhost:9092"
    #from_offsets = {topic_partion: long(offset_data["untilOffset"])}
    topic = 'new_topic'
    partition = 0
    start = 0
    topicPartion = TopicAndPartition(topic,partition)
    fromOffset = {topicPartion: start}
    #kvs = KafkaUtils.createStream(ssc, zkQuorum, "spark-streaming-consumer", {topic: 1})
    kvs = KafkaUtils.createDirectStream(ssc, [topic],kafkaParams={"metadata.broker.list": brokers},fromOffsets=fromOffset)    #print(kvs)