1. Socket： 使用 StreamingContext.socketTextStream(hostname, port) 可以从 TCP socket 接收数据流。这对于测试和调试非常方便。
from pyspark.streaming import StreamingContext
ssc = StreamingContext("local[2]", "SocketExample", 1)
lines = ssc.socketTextStream("localhost", 9999)
2. 文件系统： 使用 StreamingContext.fileStream(directory) 可以从文件系统（本地文件系统或分布式文件系统，如 HDFS）读取数据流。
from pyspark.streaming import StreamingContext
ssc = StreamingContext("local[2]", "FileStreamExample", 1)
lines = ssc.textFileStream("/path/to/directory")
3. Kafka： 使用 KafkaUtils.createDirectStream 可以从 Apache Kafka 中读取数据流。
from pyspark.streaming.kafka import KafkaUtils
from pyspark.streaming import StreamingContext
ssc = StreamingContext("local[2]", "KafkaExample", 1)
kafka_params = {"metadata.broker.list": "kafka_broker:9092"}
topics = ["topic1", "topic2"]
lines = KafkaUtils.createDirectStream(ssc, topics, kafka_params)
4. 其他输入源： Spark Streaming 还支持从其他一些输入源读取数据流，如 Flume、Twitter、HBase 等，具体取决于你的需求。
上述示例中，lines 是一个 DStream 对象，代表了输入数据流。一旦你获得了输入的 DStream，你可以对它执行各种转换和操作，例如 map、flatMap、reduceByKey 等，以处理实时数据。
转载请注明出处：http://www.zyzy.cn/article/detail/9349/Spark