编程模型
结构化数据流中的关键思想是将实时数据流视为一个不断附加的表。这导致新的流处理模型与批处理模型非常相似。您将把流式计算表示为标准批量查询，就像在静态表上一样，Spark将它作为增量查询在无界输入表上运行。让我们更详细地了解这个模型。
与kafka的集成
1、参考文档
http://spark.apache.org/docs/2.2.0/structured-streaming-kafka-integration.html
2、kafka的版本
Kafka broker version 0.10.0 or higher
3、示例1 在ide上运行
groupId = org.apache.spark
artifactId = spark-sql-kafka-0-10_2.11
version = 2.2.0
################编写代码在ide上启动##################
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder()
.master("local[2]")
.appName("StructuredStreamingKafka")
.getOrCreate()
import spark.implicits._
val df = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers","bigdata-pro02.kfk.com:9092")
.option("subscribe","weblog")
.load()
val lines = df.selectExpr("CAST(value as STRING)")//对字段进行UDF操作，并返回该列
.as[String]
val wordCount = lines.flatMap(_.split(" ")).groupBy("value").count()
//开启
val query = wordCount.writeStream
.outputMode("complete") //模式，complete，updata，
.format("console") //输出的地方在控制台
.start()
query.awaitTermination()
}
###############启动指定节点上的kafka和消息生产者##################
bin/kafka-server-start.sh config/server.properties
bin/kafka-topics.sh --create --zookeeper bigdata-pro01.kfk.com:2181,bigdata-pro02.kfk.com:2181,bigdata-pro03.kfk.com:2181 --replication-factor 3 --partitions 1 --topic weblog
3、示例2 在spark-shell上运行
jars上需要导包
kafka_2.11-0.10.0.0.jar
kafka-clients-0.10.0.0.jar
spark-sql-kafka-0-10_2.11-2.2.0.jar
spark-streaming-kafka-0-10_2.11-2.1.0.jar
#####################启动spark-shell#################
bin/spark-shell
:paste
import spark.implicits._
val df = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers","bigdata-pro02.kfk.com:9092")
.option("subscribe","weblog")
.load()
val lines = df.selectExpr("CAST(value as STRING)")
.as[String]
val wordCount = lines.flatMap(_.split(" ")).groupBy("value").count()
val query = wordCount.writeStream
.outputMode("complete")
.format("console")
.start()
query.awaitTermination()
与mysql集成——输出到mysql中
spark2.2.0暂没有api直接输出道mysql中，但是可以利用重写ForeachWriter的方法，将每一行数据写入到mysql中。如果数据量非常大，建议先写到kafka中存储，kafka按照队列的排序进行写入到mysql中
jdbcSink类
package toMysql
import java.sql._
import org.apache.spark.sql.{ForeachWriter, Row}
/**
* Created by zhongyuan on 2018/3/18.
*/
class jdbcSink(url:String,user:String,pwd:String) extends ForeachWriter[Row]{
val driver = "com.mysql.jdbc.Driver";
var statement:Statement = _;
var connection:Connection = _;
//创建连接
def open(partitionId: Long, version: Long): Boolean = {
Class.forName(driver);
connection = DriverManager.getConnection(url,user,pwd);
this.statement = connection.createStatement();
true;
}
//执行sql
override def process(value: Row): Unit = {
statement.executeUpdate("insert into wordcount values('"+value.getAs("value")+"',"+value.getAs("count")+")")
}
//关闭资源
override def close(errorOrNull: Throwable): Unit = {
connection.close()
}
}
主函数 StructuredStreamingKafkaMysql
package toMysql
import org.apache.spark.sql.{ForeachWriter, Row, SparkSession}
import org.apache.spark.sql.streaming.ProcessingTime
/**
* Created by zhongyuan on 2018/3/18.
*/
object StructuredStreamingKafkaMysql {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder()
.master("local[2]")
.appName("StructuredStreamingKafkaMysql")
.getOrCreate()
import spark.implicits._
val df = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers","bigdata-pro02.kfk.com:9092")
.option("subscribe","weblog")
.load()
val lines = df.selectExpr("CAST(value as STRING)")//对字段进行UDF操作，并返回该列
.as[String]
val wordCount = lines.flatMap(_.split(" ")).groupBy("value").count()
//输出到外部mysql
val url = "jdbc:mysql://bigdata-pro03.kfk.com/spark"
val user = "root"
val pwd = "123456"
val writer:ForeachWriter[Row] = new jdbcSink(url,user,pwd);//新建自定义类
val query = wordCount
.writeStream
.foreach(writer)//forEach()里只能写ForeachWriter[Row]类，所以需要指定writer的类型
.outputMode("update")
.trigger(ProcessingTime("25 seconds"))
.start()
query.awaitTermination()
}
}
执行顺序
先启动指定所有节点的zookeeper
在启动指定节点的kafka
启动指定节点的topic为weblog的消息生产者
启动指定节点的mysql
启动ide程序
利用消息producer来发送消息
查询mysql中是否有数据
