### Pyspark Streaming 使用不同输入源 示例代码 PySpark Streaming 支持多种不同的输入源，这些输入源可以来自文件系统、套接字连接或其他外部系统。以下是几个常见的 PySpark Streaming 输入源及其对应的示例代码。 #### 1. 基于 Socket 的输入流 通过 `socketTextStream` 方法可以从网络上的某个端口读取数据流。这是一个简单的例子： ```python from pyspark import SparkConf, SparkContext from pyspark.streaming import StreamingContext # 配置 Spark Conf 和 Streaming Context conf = SparkConf().setMaster("local[2]").setAppName("SocketWordCount") sc = SparkContext.getOrCreate(conf) ssc = StreamingContext(sc, batchDuration=5) # 连接到 localhost 上的 9999 端口获取数据 lines = ssc.socketTextStream("localhost", 9999) # 对每一行进行单词分割并统计频率 words = lines.flatMap(lambda line: line.split(" ")) pairs = words.map(lambda word: (word, 1)) wordCounts = pairs.reduceByKey(lambda x, y: x + y) # 打印结果 wordCounts.pprint() ssc.start() ssc.awaitTermination() ``` 此代码片段展示了如何从指定 IP 地址和端口号接收实时文本数据，并对其进行简单处理[^3]。 --- #### 2. 文件作为输入源 可以通过监控目录的方式加载新到达的文件内容。每当有新的文件被放置到目标路径下时，它们会被自动检测并加入到 DStream 中。 ```python from pyspark import SparkConf, SparkContext from pyspark.streaming import StreamingContext # 初始化 Spark Conf 和 Streaming Context conf = SparkConf().setMaster("local[2]").setAppName("FileInputExample") sc = SparkContext.getOrCreate(conf) ssc = StreamingContext(sc, batchInterval=10) # 定义要监听的文件夹位置 data_stream = ssc.textFileStream("/path/to/your/input/folder") # 处理逻辑：比如按空格拆分字符串计数 result = data_stream.flatMap(lambda line: line.split(' ')).map(lambda w: (w, 1)).reduceByKey(lambda v1, v2: v1+v2) # 输出前几条记录查看效果 result.pprint() ssc.start() ssc.awaitTermination() ``` 这里演示了当某些文件写入特定目录后触发的操作流程[^2]。 --- #### 3. Kafka 数据源集成 Kafka 是一种高吞吐量的消息队列服务，常与 Spark Streaming 结合使用以实现高效的大规模消息消费场景。 首先需要安装额外依赖项： ```bash pip install kafka-python ``` 接着编写 Python 脚本如下所示： ```python from pyspark import SparkConf, SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.kafka import KafkaUtils def createContext(): conf = SparkConf().setMaster("local[*]").setAppName("DirectKafkaWordCount") sc = SparkContext.getOrCreate(conf) # 创建 SreamingContext 实例 ssc = StreamingContext(sc, batchSizeInSeconds=2) brokers, topic = "localhost:9092", ["test"] kvs = KafkaUtils.createDirectStream(ssc, topics=[topic], kafkaParams={"metadata.broker.list": brokers}) lines = kvs.map(lambda x: x[1]) counts = lines.flatMap(lambda line: line.split(" ")) \ .map(lambda word: (word, 1)) \ .reduceByKey(lambda a, b: a+b) counts.pprint() return ssc if __name__ == "__main__": checkpointDir = "/tmp/spark-streaming-checkpoint" ssc = StreamingContext.getActiveOrCreate(checkpointDir, lambda: createContext()) ssc.start() ssc.awaitTermination() ``` 该脚本利用 Direct API 方式直接对接 Apache Kafka 主题订阅机制完成任务需求[^4]。 --- ### 总结 以上分别介绍了基于 socket 流、文件系统以及 Kafka 消息中间件三种典型方式接入 PySpark Streaming 平台的具体方法论和技术细节说明文档链接地址分别为[^1],[^2],[^3].