【Jacek Laskowski】Spark Structured Streaming

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

1. Spark Structured Streaming
   
2. 
   
3. Welcome to Spark Structured Streaming notebook!
   
4. 
   
5. I’m Jacek Laskowski, an independent consultant who is passionate about Apache Spark, Apache Kafka, Scala and sbt (with some flavour of Apache Mesos, Hadoop YARN, and quite recently DC/OS). I lead Warsaw Scala Enthusiasts and Warsaw Spark meetups in Warsaw, Poland.
   
6. 
   
7. Contact me at jacek@japila.pl or @jaceklaskowski to discuss Apache Spark opportunities, e.g. courses, workshops, mentoring or application development services.
   
8. 
   
9. If you like the Apache Spark notes you should seriously consider participating in my own, very hands-on Spark Workshops.
   
10. 
    
11. Tip
    
12. I’m also writing Mastering Apache Spark 2 Notebook, Apache Kafka Notebook and Spark Streaming Notebook.
    
13. Spark Structured Streaming Notebook serves as the ultimate place of mine to collect all the nuts and bolts of using Spark Structured Streaming. The notes aim to help me designing and developing better products with Apache Spark. It is also a viable proof of my understanding of Apache Spark. I do eventually want to reach the highest level of mastery in Apache Spark (as do you!)
    
14. 
    
15. The collection of notes serves as the study material for my trainings, workshops, videos and courses about Apache Spark. Follow me on twitter @jaceklaskowski to know it early. You will also learn about the upcoming events about Apache Spark.
    
16. 
    
17. Expect text and code snippets from Spark’s mailing lists, the official documentation of Apache Spark, StackOverflow, blog posts, books from O’Reilly (and other publishers), press releases, conferences, YouTube or Vimeo videos, Quora, the source code of Apache Spark, etc. Attribution follows.

复制代码

本帖隐藏的内容

Spark Structured Streaming Jacek Laskowski.pdf (421.27 KB)

2017-5-15 07:35:10 上传

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏1 回帖

关键词：Ask WSK Apache Spark Conferences Independent

1. DataStreamReader
   
2. 
   
3. DataStreamReader is an interface for reading streaming data in DataFrame from data sources with specified format, schema and options.
   
4. DataStreamReader offers support for the built-in formats: json, csv, parquet, text. parquet format is the default data source as configured using spark.sql.sources.default setting.
   
5. DataStreamReader is available using SparkSession.readStream method.
   
6. val spark: SparkSession = ...
   
7. 
   
8. val schema = spark.read
   
9.   .format("csv")
   
10.   .option("header", true)
    
11.   .option("inferSchema", true)
    
12.   .load("csv-logs/*.csv")
    
13.   .schema
    
14. 
    
15. val df = spark.readStream
    
16.   .format("csv")
    
17.   .schema(schema)
    
18.   .load("csv-logs/*.csv")

复制代码

1. FileStreamSource
   
2. 
   
3. FileStreamSource is a Source that reads text files from path directory as they appear. It uses LongOffset offsets.+
   
4. 
   
5. Note
   
6. It is used by DataSource.createSource for FileFormat.
   
7. You can provide the schema of the data and dataFrameBuilder - the function to build a DataFrame in getBatch at instantiation time.
   
8. // NOTE The source directory must exist
   
9. // mkdir text-logs
   
10. 
    
11. val df = spark.readStream
    
12.   .format("text")
    
13.   .option("maxFilesPerTrigger", 1)
    
14.   .load("text-logs")
    
15. 
    
16. scala> df.printSchema
    
17. root
    
18. |-- value: string (nullable = true)

复制代码

1. Creating KafkaSourceRDD Instance
   
2. 
   
3. KafkaSourceRDD takes the following when created:
   
4. SparkContext
   
5. Collection of key-value settings for executors reading records from Kafka topics
   
6. Collection of KafkaSourceRDDOffsetRange offsets
   
7. Timeout (in milliseconds) to poll data from Kafka
   
8. Used when KafkaSourceRDD is requested for records (for given offsets) and in turn requests CachedKafkaConsumer to poll for Kafka’s ConsumerRecords.
   
9. Flag to…FIXME
   
10. Flag to…FIXME
    
11. KafkaSourceRDD initializes the internal registries and counters.

复制代码

MemoryStream

1. val spark: SparkSession = ???
   
2. 
   
3. implicit val ctx = spark.sqlContext
   
4. 
   
5. import org.apache.spark.sql.execution.streaming.MemoryStream
   
6. // It uses two implicits: Encoder[Int] and SQLContext
   
7. val intsIn = MemoryStream[Int]
   
8. 
   
9. val ints = intsIn.toDF
   
10.   .withColumn("t", current_timestamp())
    
11.   .withWatermark("t", "5 minutes")
    
12.   .groupBy(window($"t", "5 minutes") as "window")
    
13.   .agg(count("*") as "total")
    
14. 
    
15. import org.apache.spark.sql.streaming.OutputMode
    
16. import org.apache.spark.sql.streaming.ProcessingTime
    
17. import scala.concurrent.duration._
    
18. val totalsOver5mins = ints.writeStream
    
19.   .format("memory")
    
20.   .queryName("totalsOver5mins")
    
21.   .outputMode(OutputMode.Append)
    
22.   .trigger(ProcessingTime(30.seconds))
    
23.   .start
    
24. 
    
25. scala> val zeroOffset = intsIn.addData(0, 1, 2)
    
26. zeroOffset: org.apache.spark.sql.execution.streaming.Offset = #0
    
27. 
    
28. totalsOver5mins.processAllAvailable()
    
29. spark.table("totalsOver5mins").show
    
30. 
    
31. scala> intsOut.show

复制代码

TextSocketSource

TextSocketSource is a streaming source that reads lines from a socket at the host and port (defined by parameters).+

It uses lines internal in-memory buffer to keep all of the lines that were read from a socket forever.

1. import org.apache.spark.sql.SparkSession
   
2. val spark: SparkSession = SparkSession.builder.getOrCreate()
   
3. 
   
4. // Connect to localhost:9999
   
5. // You can use "nc -lk 9999" for demos
   
6. val textSocket = spark.readStream
   
7.   .format("socket")
   
8.   .option("host", "localhost")
   
9.   .option("port", 9999)
   
10.   .load
    
11. 
    
12. import org.apache.spark.sql.Dataset
    
13. val lines: Dataset[String] = textSocket.as[String].map(_.toUpperCase)
    
14. 
    
15. val query = lines.writeStream.format("console").start
    
16. 
    
17. // Start typing the lines in nc session
    
18. // They will appear UPPERCASE in the terminal

复制代码

1. DataStreamWriter
   
2. 
   
3. DataStreamWriter is a part of Structured Streaming that is responsible for writing the output of streaming queries to sinks and therefore starting their execution.
   
4. val people: Dataset[Person] = ...
   
5. 
   
6. import org.apache.spark.sql.streaming.ProcessingTime
   
7. import scala.concurrent.duration._
   
8. import org.apache.spark.sql.streaming.OutputMode.Complete
   
9. df.writeStream
   
10.   .queryName("textStream")
    
11.   .outputMode(Complete)
    
12.   .trigger(ProcessingTime(10.seconds))
    
13.   .format("console")
    
14.   .start

复制代码

1. ConsoleSink
   
2. 
   
3. ConsoleSink is a streaming sink that is registered as the console format.
   
4. val spark: SparkSession = ...
   
5. spark.readStream
   
6.   .format("text")
   
7.   .load("server-logs/*.out")
   
8.   .as[String]
   
9.   .writeStream
   
10.   .queryName("server-logs processor")
    
11.   .format("console")  // <-- uses ConsoleSink
    
12.   .start
    
13. 
    
14. scala> spark.streams.active.foreach(println)
    
15. Streaming Query - server-logs processor [state = ACTIVE]
    
16. 
    
17. // in another terminal
    
18. $ echo hello > server-logs/hello.out

复制代码

1. FileStreamSink
   
2. 
   
3. FileStreamSink is the streaming sink for the parquet format.
   
4. 
   
5. import scala.concurrent.duration._
   
6. import org.apache.spark.sql.streaming.{OutputMode, ProcessingTime}
   
7. val out = in.writeStream
   
8.   .format("parquet")
   
9.   .option("path", "parquet-output-dir")
   
10.   .option("checkpointLocation", "checkpoint-dir")
    
11.   .trigger(ProcessingTime(5.seconds))
    
12.   .outputMode(OutputMode.Append)
    
13.   .start()
    
14. FileStreamSink supports Append output mode only.
    
15. It uses spark.sql.streaming.fileSink.log.deletion (as isDeletingExpiredLog)

复制代码

1. ForeachSink
   
2. 
   
3. ForeachSink is a typed Sink that passes records (of the type T) to ForeachWriter (one record at a time per partition).
   
4. It is used exclusively in foreach operator.
   
5. val records = spark.readStream
   
6.   .format("text")
   
7.   .load("server-logs/*.out")
   
8.   .as[String]
   
9. 
   
10. import org.apache.spark.sql.ForeachWriter
    
11. val writer = new ForeachWriter[String] {
    
12.   override def open(partitionId: Long, version: Long) = true
    
13.   override def process(value: String) = println(value)
    
14.   override def close(errorOrNull: Throwable) = {}
    
15. }
    
16. 
    
17. records.writeStream
    
18.   .queryName("server-logs processor")
    
19.   .foreach(writer)
    
20.   .start

复制代码