摸鱼大数据——Spark Structured Steaming——结构化流的编程综合案例

7、综合案例

需求: 已知文件中存储了多个单词,要求计算统计出现的次数

7.1 词频统计_读取文件方式

 # 导包
 import os
 from pyspark.sql import SparkSession,functions as F
 ​
 # 绑定指定的python解释器
 os.environ['SPARK_HOME'] = '/export/server/spark'
 os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
 os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
 ​
 # 创建main函数
 if __name__ == '__main__':
     # 1.创建SparkContext对象
     spark = SparkSession.builder\
         .config('spark.sql.shuffle.partitions',1)\
         .appName('pyspark_demo')\
         .master('local[*]')\
         .getOrCreate()
 ​
     # 2.数据输入
     # 注意: 路径必须是目录路径,因为readStream会自动读取此目录下的所有文件,有新增会触发接着读
     df = spark.readStream\
         .format('text')\
         .load('file:///export/data/spark_project/structured_Streaming/data/')
 ​
     # 查看数据类型
     print(type(df))
     # 3.数据处理(切分,转换,分组聚合)
     # 和SparkSQL操作一模一样,支持sql和dsl两种风格
     # SQL方式
     df.createTempView('tb')
     sql_df = spark.sql("""
         select words,count(1) as cnt
         from (
           select explode(split(value,' ')) as words from tb
         ) t group by words
     """)
     # DSL方式
     dsl_df = df.select(
         F.explode(F.split('value',' ')).alias('words')
     ).groupBy('words').agg(
         F.count('words').alias('cnt')
     )
 ​
     # 4.数据输出
     # 注意: 输出不能使用原来sparksql的show()
     # 注意: 如果需要多开启多个输出,.awaitTermination()只需要在最后一个出现即可
     sql_df.writeStream.format('console').outputMode('complete').start()
     dsl_df.writeStream.format('console').outputMode('complete').start().awaitTermination()
     # 5.关闭资源
     spark.stop()

7.2 词频统计_Socket方式

 # 导包
 import os
 from pyspark.sql import SparkSession,functions as F
 ​
 # 绑定指定的python解释器
 os.environ['SPARK_HOME'] = '/export/server/spark'
 os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
 os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'
 ​
 # 创建main函数
 if __name__ == '__main__':
     # 1.创建SparkContext对象
     spark = SparkSession.builder\
         .config('spark.sql.shuffle.partitions',1)\
         .appName('pyspark_demo')\
         .master('local[*]')\
         .getOrCreate()
 ​
     # 2.数据输入
     # 注意: 路径必须是目录路径,因为readStream会自动读取此目录下的所有文件,有新增会触发接着读
     df = spark.readStream\
         .format('socket')\
         .option('host',"192.168.88.161")\
         .option('port',"55555")\
         .load()
 ​
     # 查看数据类型
     print(type(df))
     # 3.数据处理(切分,转换,分组聚合)
     # 和SparkSQL操作一模一样,支持sql和dsl两种风格
     # SQL方式
     df.createTempView('tb')
     sql_df = spark.sql("""
         select words,count(1) as cnt
         from (
           select explode(split(value,' ')) as words from tb
         ) t group by words
     """)
     # DSL方式
     dsl_df = df.select(
         F.explode(F.split('value',' ')).alias('words')
     ).groupBy('words').agg(
         F.count('words').alias('cnt')
     )
 ​
     # 4.数据输出
     # 注意: 输出不能使用原来sparksql的show()
     # 注意: 如果需要多开启多个输出,.awaitTermination()只需要在最后一个出现即可
     sql_df.writeStream.format('console').outputMode('complete').start()
     dsl_df.writeStream.format('console').outputMode('complete').start().awaitTermination()
     # 5.关闭资源
     spark.stop()
 ​

7.3 自动生成数据_Rate方式

 from pyspark.sql import SparkSession
 import os
 ​
 os.environ["SPARK_HOME"] = "/export/server/spark"
 os.environ["PYSPARK_PYTHON"] = "/root/anaconda3/bin/python"
 os.environ["PYSPARK_DRIVER_PYTHON"] = "/root/anaconda3/bin/python"
 ​
 if __name__ == '__main__':
     # 1.创建SparkSession对象
     spark = SparkSession.builder \
         .appName("StructuredStream_rate") \
         .master('local[*]') \
         .getOrCreate()
         
     # 2。读取数据
     df = spark.readStream \
         .format('rate') \
         .option("rowsPerSecond", "5") \
         .option('numPartitions', 1) \
         .load()
 ​
     # 3.数据处理
     # 略
 ​
     # 4.数据输出:
     df.writeStream \
         .format('console') \
         .outputMode('update') \
         .option('truncate', 'false') \
         .start() \
         .awaitTermination()
 ​
     # 5.关闭资源
     spark.stop()