from __future__ import print_function import sys from pyspark import SparkContext from pyspark.streaming import StreamingContext from pyspark.streaming.flume import FlumeUtils import pyspark if __name__ == "__main__": if len(sys.argv) != 3: print("Usage: flume_wordcount.py <hostname> <port>", file=sys.stderr) exit(-1) sc = SparkContext(appName="FlumeEventCount") ssc = StreamingContext(sc, 2) hostname= sys.argv[1] port = int(sys.argv[2]) stream = FlumeUtils.createStream(ssc, hostname, port,pyspark.StorageLevel.MEMORY_AND_DISK_SER_2) stream.count().map(lambda cnt : "Recieve " + str(cnt) +" Flume events!!!!").pprint() ssc.start() ssc.awaitTermination()

这是一个使用 PySpark Streaming 监听 Flume 事件的示例代码，具体解释如下：

- `from __future__ import print_function`：导入 print 函数，因为在 Python 2.x 中 print 是关键字，而在 Python 3.x 中是函数。
- `import sys`：导入 sys 模块，用于获取命令行参数。
- `from pyspark import SparkContext`：导入 SparkContext 类，用于创建 SparkContext 对象。
- `from pyspark.streaming import StreamingContext`：导入 StreamingContext 类，用于创建 StreamingContext 对象。
- `from pyspark.streaming.flume import FlumeUtils`：导入 FlumeUtils 类，用于创建 Flume 事件流。
- `import pyspark`：导入 pyspark 模块，用于设置存储级别。
- `if __name__ == "__main__":`：如果当前脚本为主函数，则执行下面的代码。
- `if len(sys.argv) != 3:`：如果命令行参数不等于 3，则输出错误信息并退出。
- `sc = SparkContext(appName="FlumeEventCount")`：创建 SparkContext 对象，设置应用程序名称为 "FlumeEventCount"。
- `ssc = StreamingContext(sc, 2)`：创建 StreamingContext 对象，设置批处理间隔为 2 秒。
- `hostname= sys.argv[1]`：获取命令行参数中的主机名。
- `port = int(sys.argv[2])`：获取命令行参数中的端口号，并将其转换为整数。
- `stream = FlumeUtils.createStream(ssc, hostname, port,pyspark.StorageLevel.MEMORY_AND_DISK_SER_2)`：创建 Flume 事件流，使用 MEMORY_AND_DISK_SER_2 存储级别。
- `stream.count().map(lambda cnt : "Recieve " + str(cnt) +" Flume events!!!!").pprint()`：对事件流进行计数，并将计数结果转换为字符串输出。
- `ssc.start()`：启动 StreamingContext。
- `ssc.awaitTermination()`：等待 StreamingContext 结束。