Spark笔记17-Structured Streaming

# StructuredNetWordCount.py

from pyspark.sql import SparkSession
from pyspark.sql.functions import split
from pyspark.sql.functions import explode

# 创建SparkSession对象
if __name__ == "__main__":
  spark = SparkSession.builder.appName("StructuredNetworkCount").getOrCreate()
  spark.sparkContext.setLogLevel("WARN")

# 创建输入数据源
lines = spark.readStream.formaat("socket").option("host", "localhost").option("port", 9999).load()

# 定义流计算过程
words = lines.select(explode(split(lines.value, " ")).alias("word"))
wordsCounts = words.groupBy("word").count()

# 启动流计算并且输出结果
query = wordCounts.writeStream.outputMode("complete").format("console").trigger(processingTime="8 seconds")
.start()  # complete 表示输出模式

query.awaitTermination()

# 启动HDFS
cd /usr/local/hadoop
sbin/start-dfs.sh

# 新建数据源终端
nc -lk 9999   # 启动服务端；需要输入语句

# 新建流计算终端：客户端
cd /usr/local/spark/mycode/structuredstreaming/
/usr/local/spark/bin/spark-submit StructuredNetWordCount.py