| 编程模型 | 解释 |
|---|---|
|
|
|
|
|
|
|
|
|
RDD 的优点
-
面向对象的操作方式
-
可以处理任何类型的数据
RDD 的缺点
-
运行速度比较慢, 执行过程没有优化
-
API比较僵硬, 对结构化数据的访问和操作没有优化
DataFrame 的优点
-
针对结构化数据高度优化, 可以通过列名访问和转换数据
-
增加
Catalyst优化器, 执行过程是优化的, 避免了因为开发者的原因影响效率
DataFrame 的缺点
-
只能操作结构化数据
-
只有无类型的
API, 也就是只能针对列和SQL操作数据,API依然僵硬
Dataset 的优点
-
结合了
RDD和DataFrame的API, 既可以操作结构化数据, 也可以操作非结构化数据 -
既有有类型的
API也有无类型的API, 灵活选择
整体结构
-
Socket Server等待Structured Streaming程序连接 -
Structured Streaming程序启动, 连接Socket Server, 等待Socket Server发送数据 -
Socket Server发送数据,Structured Streaming程序接收数据 -
Structured Streaming程序接收到数据后处理数据 -
数据处理后, 生成对应的结果集, 在控制台打印
HDFS读数据：
目标
-
在数据处理的时候, 经常会遇到这样的场景
-
有时候也会遇到这样的场景
-
以上两种场景有两个共同的特点
-
会产生大量小文件在
HDFS上 -
数据需要处理
-
-
通过本章节的学习, 便能够更深刻的理解这种结构, 具有使用
Structured Streaming整合HDFS, 从其中读取数据的能力
案例流程
-
编写
Python小程序, 在某个目录生成大量小文件-
Python是解释型语言, 其程序可以直接使用命令运行无需编译, 所以适合编写快速使用的程序, 很多时候也使用Python代替Shell -
使用
Python程序创建新的文件, 并且固定的生成一段JSON文本写入文件 -
在真实的环境中, 数据也是一样的不断产生并且被放入
HDFS中, 但是在真实场景下, 可能是Flume把小文件不断上传到HDFS中, 也可能是Sqoop增量更新不断在某个目录中上传小文件
-
-
使用
Structured Streaming汇总数据-
HDFS中的数据是不断的产生的, 所以也是流式的数据 -
数据集是
JSON格式, 要有解析JSON的能力 -
因为数据是重复的, 要对全局的流数据进行汇总和去重, 其实真实场景下的数据清洗大部分情况下也是要去重的
-
-
使用控制台展示数据
-
最终的数据结果以表的形式呈现
-
使用控制台展示数据意味着不需要在修改展示数据的代码, 将
Sink部分的内容放在下一个大章节去说明 -
真实的工作中, 可能数据是要落地到
MySQL,HBase,HDFS这样的存储系统中
-
难点和易错点
import os for index in range(100): content = """ {"name": "Michael"} {"name": "Andy", "age": 30} {"name": "Justin", "age": 19} """ file_name = "/export/dataset/text{0}.json".format(index) with open(file_name, "w") as file: file.write(content) os.system("/export/servers/hadoop/bin/hdfs dfs -mkdir -p /dataset/dataset/") os.system("/export/servers/hadoop/bin/hdfs dfs -put {0} /dataset/dataset/".format(file_name))
执行python文件
删除hdfs目录
package cn.itcast.structured import org.apache.spark.sql.SparkSession import org.apache.spark.sql.streaming.OutputMode import org.apache.spark.sql.types.StructType object HDFSSource { def main(args: Array[String]): Unit = { System.setProperty("hadoop.home.dir", "C:\winutil") // 1. 创建 SparkSession val spark = SparkSession.builder() .appName("hdfs_source") .master("local[6]") .getOrCreate() // 2. 数据读取, 目录只能是文件夹, 不能是某一个文件 val schema = new StructType() .add("name", "string") .add("age", "integer") val source = spark.readStream .schema(schema) .json("hdfs://node01:8020/dataset/dataset") // 3. 输出结果 source.writeStream .outputMode(OutputMode.Append()) .format("console") .start() .awaitTermination() } }
运行 Python 程序
kafka：
Kafka 的特点
Topic 和 Partitions
-
消息和事件经常是不同类型的, 例如用户注册是一种消息, 订单创建也是一种消息
-
Kafka中使用Topic来组织不同类型的消息
-
Kafka中的Topic要承受非常大的吞吐量, 所以Topic应该是可以分片的, 应该是分布式的
总结
创建 Topic 并输入数据到 Topic
使用 Spark 读取 Kafka 的 Topic
sink
从 Kafka 中获取数据, 简单处理, 再次放入 Kafka
package cn.itcast.structured import org.apache.spark.sql.{DataFrame, Dataset, SparkSession} object HDFSSink { def main(args: Array[String]): Unit = { System.setProperty("hadoop.home.dir", "C:\winutil") // 1. 创建 SparkSession val spark = SparkSession.builder() .appName("hdfs_sink") .master("local[6]") .getOrCreate() import spark.implicits._ // 2. 读取 Kafka 数据 val source: Dataset[String] = spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "node01:9092,node02:9092,node03:9092") .option("subscribe", "streaming_test_2") .option("startingOffsets", "earliest") .load() .selectExpr("CAST(value AS STRING) as value") .as[String] // 1::Toy Story (1995)::Animation|Children's|Comedy // 3. 处理 CSV, Dataset(String), Dataset(id, name, category) val result = source.map(item => { val arr = item.split("::") (arr(0).toInt, arr(1).toString, arr(2).toString) }).as[(Int, String, String)].toDF("id", "name", "category") // 4. 落地到 HDFS 中 result.writeStream .format("parquet") .option("path", "dataset/streaming/moives/") .option("checkpointLocation", "checkpoint") .start() .awaitTermination() } }
掌握 Foreach 模式理解如何扩展 Structured Streaming 的 Sink, 同时能够将数据落地到 MySQL
package cn.itcast.structured import java.sql.{Connection, DriverManager, Statement} import org.apache.spark.sql.{Dataset, ForeachWriter, Row, SparkSession} object ForeachSink { def main(args: Array[String]): Unit = { System.setProperty("hadoop.home.dir", "C:\winutil") // 1. 创建 SparkSession val spark = SparkSession.builder() .appName("hdfs_sink") .master("local[6]") .getOrCreate() import spark.implicits._ // 2. 读取 Kafka 数据 val source: Dataset[String] = spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "node01:9092,node02:9092,node03:9092") .option("subscribe", "streaming_test_2") .option("startingOffsets", "earliest") .load() .selectExpr("CAST(value AS STRING) as value") .as[String] // 1::Toy Story (1995)::Animation|Children's|Comedy // 3. 处理 CSV, Dataset(String), Dataset(id, name, category) val result = source.map(item => { val arr = item.split("::") (arr(0).toInt, arr(1).toString, arr(2).toString) }).as[(Int, String, String)].toDF("id", "name", "category") // 4. 落地到 MySQL class MySQLWriter extends ForeachWriter[Row] { private val driver = "com.mysql.jdbc.Driver" private var connection: Connection = _ private val url = "jdbc::mysql://node01:3306/streaming-movies-result" private var statement: Statement = _ override def open(partitionId: Long, version: Long): Boolean = { Class.forName(driver) connection = DriverManager.getConnection(url) statement = connection.createStatement() true } override def process(value: Row): Unit = { statement.executeUpdate(s"insert into movies values(${value.get(0)}, ${value.get(1)}, ${value.get(2)})") } override def close(errorOrNull: Throwable): Unit = { connection.close() } } result.writeStream .foreach(new MySQLWriter) .start() .awaitTermination() } }
triggers
package cn.itcast.structured import org.apache.spark.sql.SparkSession import org.apache.spark.sql.streaming.{OutputMode, Trigger} object Triggers { def main(args: Array[String]): Unit = { System.setProperty("hadoop.home.dir", "C:\winutil") // 创建数据源 val spark = SparkSession.builder() .appName("triggers") .master("local[6]") .getOrCreate() spark.sparkContext.setLogLevel("WARN") // timestamp, value val source = spark.readStream .format("rate") .load() // 简单处理 // val result = source // 落地 source.writeStream .format("console") .outputMode(OutputMode.Append()) .trigger(Trigger.Once()) .start() .awaitTermination() } }