Spark Structured Streaming HelloWorld

Preface

Spark Structured Streaming+Kafka+Hbase Scala Version tutorial , The whole entrance .

Text

1.Spark Version selection

Select the corresponding version of your server ; Document address :
https://spark.apache.org/docs/
This address is opened with the version number , Choose your own environment Spark That's all right. ;
So here I'm going to use theta 2.4.5; The latest version of the document is 3.3.3

2. The official example

After entering the corresponding version, you can find Spark The main function of , Here's the picture
Spark Streaming It has been clearly marked as old API 了 , new API Namely Structured Streaming, The picture is circled in red , So what I am currently using is new API.Structured Streaming

HelloWorld Code

The official one is simple word count Example

// Create DataFrame representing the stream of input lines from connection to localhost:9999
val lines = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

// Split the lines into words
val words = lines.as[String].flatMap(_.split(" "))

// Generate running word count
val wordCounts = words.groupBy("value").count()

Batch code example

The official example , Here is my understanding ,streamingDF Is the data of a batch ;foreachBatch Is to cycle each batch ; The data in the batch is batchDF, Print batch number batchId You can see that this batch number is a self increasing number ;

streamingDF.writeStream.foreachBatch {
     (batchDF: DataFrame, batchId: Long) =>
	// This line is for caching , In this way, subsequent operations will not repeat the previous transform Operation 
  batchDF.persist()
  // Operate the data in a batch , The specific basis is what operation is written differently 
  batchDF.write.format(...).save(...)  // location 1
  batchDF.write.format(...).save(...)  // location 2
  // The cache must be released after completion 
  batchDF.unpersist()
}