-
Notifications
You must be signed in to change notification settings - Fork 7
/
Copy pathstreaming-programming-guide.html
executable file
·2669 lines (2163 loc) · 211 KB
/
streaming-programming-guide.html
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
<!DOCTYPE html>
<!--[if lt IE 7]> <html class="no-js lt-ie9 lt-ie8 lt-ie7"> <![endif]-->
<!--[if IE 7]> <html class="no-js lt-ie9 lt-ie8"> <![endif]-->
<!--[if IE 8]> <html class="no-js lt-ie9"> <![endif]-->
<!--[if gt IE 8]><!--> <html class="no-js"> <!--<![endif]-->
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">
<title>Spark Streaming - Spark 2.0.0 Documentation</title>
<meta name="description" content="Spark Streaming programming guide and tutorial for Spark 2.0.0">
<link rel="stylesheet" href="css/bootstrap.min.css">
<style>
body {
padding-top: 60px;
padding-bottom: 40px;
}
</style>
<meta name="viewport" content="width=device-width">
<link rel="stylesheet" href="css/bootstrap-responsive.min.css">
<link rel="stylesheet" href="css/main.css">
<script src="js/vendor/modernizr-2.6.1-respond-1.1.0.min.js"></script>
<link rel="stylesheet" href="css/pygments-default.css">
</head>
<body>
<!--[if lt IE 7]>
<p class="chromeframe">You are using an outdated browser. <a href="http://browsehappy.com/">Upgrade your browser today</a> or <a href="http://www.google.com/chromeframe/?redirect=true">install Google Chrome Frame</a> to better experience this site.</p>
<![endif]-->
<!-- This code is taken from http://twitter.github.com/bootstrap/examples/hero.html -->
<div class="navbar navbar-fixed-top" id="topbar">
<div class="navbar-inner">
<div class="container">
<div class="brand"><a href="index.html">
<img src="img/spark-logo-hd.png" style="height:50px;"/></a><span class="version">2.0.0</span>
</div>
<ul class="nav">
<!--TODO(andyk): Add class="active" attribute to li some how.-->
<li><a href="index.html">Overview</a></li>
<li class="dropdown">
<a href="#" class="dropdown-toggle" data-toggle="dropdown">Programming Guides<b class="caret"></b></a>
<ul class="dropdown-menu">
<li><a href="quick-start.html">Quick Start</a></li>
<li><a href="programming-guide.html">Spark Programming Guide</a></li>
<li class="divider"></li>
<li><a href="streaming-programming-guide.html">Spark Streaming</a></li>
<li><a href="sql-programming-guide.html">DataFrames, Datasets and SQL</a></li>
<li><a href="mllib-guide.html">MLlib (Machine Learning)</a></li>
<li><a href="graphx-programming-guide.html">GraphX (Graph Processing)</a></li>
<li><a href="sparkr.html">SparkR (R on Spark)</a></li>
</ul>
</li>
<li class="dropdown">
<a href="#" class="dropdown-toggle" data-toggle="dropdown">API Docs<b class="caret"></b></a>
<ul class="dropdown-menu">
<li><a href="api/scala/index.html#org.apache.spark.package">Scala</a></li>
<li><a href="api/java/index.html">Java</a></li>
<li><a href="api/python/index.html">Python</a></li>
<li><a href="api/R/index.html">R</a></li>
</ul>
</li>
<li class="dropdown">
<a href="#" class="dropdown-toggle" data-toggle="dropdown">Deploying<b class="caret"></b></a>
<ul class="dropdown-menu">
<li><a href="cluster-overview.html">Overview</a></li>
<li><a href="submitting-applications.html">Submitting Applications</a></li>
<li class="divider"></li>
<li><a href="spark-standalone.html">Spark Standalone</a></li>
<li><a href="running-on-mesos.html">Mesos</a></li>
<li><a href="running-on-yarn.html">YARN</a></li>
</ul>
</li>
<li class="dropdown">
<a href="api.html" class="dropdown-toggle" data-toggle="dropdown">More<b class="caret"></b></a>
<ul class="dropdown-menu">
<li><a href="configuration.html">Configuration</a></li>
<li><a href="monitoring.html">Monitoring</a></li>
<li><a href="tuning.html">Tuning Guide</a></li>
<li><a href="job-scheduling.html">Job Scheduling</a></li>
<li><a href="security.html">Security</a></li>
<li><a href="hardware-provisioning.html">Hardware Provisioning</a></li>
<li class="divider"></li>
<li><a href="building-spark.html">Building Spark</a></li>
<li><a href="https://cwiki.apache.org/confluence/display/SPARK/Contributing+to+Spark">Contributing to Spark</a></li>
<li><a href="https://cwiki.apache.org/confluence/display/SPARK/Supplemental+Spark+Projects">Supplemental Projects</a></li>
</ul>
</li>
</ul>
<!--<p class="navbar-text pull-right"><span class="version-text">v2.0.0</span></p>-->
</div>
</div>
</div>
<div class="container-wrapper">
<div class="content" id="content">
<h1 class="title">Spark Streaming Programming Guide</h1>
<ul id="markdown-toc">
<li><a href="#overview">Overview</a></li>
<li><a href="#a-quick-example">A Quick Example</a></li>
<li><a href="#basic-concepts">Basic Concepts</a> <ul>
<li><a href="#linking">Linking</a></li>
<li><a href="#initializing-streamingcontext">Initializing StreamingContext</a></li>
<li><a href="#discretized-streams-dstreams">Discretized Streams (DStreams)</a></li>
<li><a href="#input-dstreams-and-receivers">Input DStreams and Receivers</a></li>
<li><a href="#transformations-on-dstreams">Transformations on DStreams</a></li>
<li><a href="#output-operations-on-dstreams">Output Operations on DStreams</a></li>
<li><a href="#accumulators-and-broadcast-variables">Accumulators and Broadcast Variables</a></li>
<li><a href="#dataframe-and-sql-operations">DataFrame and SQL Operations</a></li>
<li><a href="#mllib-operations">MLlib Operations</a></li>
<li><a href="#caching--persistence">Caching / Persistence</a></li>
<li><a href="#checkpointing">Checkpointing</a></li>
<li><a href="#deploying-applications">Deploying Applications</a></li>
<li><a href="#monitoring-applications">Monitoring Applications</a></li>
</ul>
</li>
<li><a href="#performance-tuning">Performance Tuning</a> <ul>
<li><a href="#reducing-the-batch-processing-times">Reducing the Batch Processing Times</a></li>
<li><a href="#setting-the-right-batch-interval">Setting the Right Batch Interval</a></li>
<li><a href="#memory-tuning">Memory Tuning</a></li>
</ul>
</li>
<li><a href="#fault-tolerance-semantics">Fault-tolerance Semantics</a></li>
<li><a href="#migration-guide-from-091-or-below-to-1x">Migration Guide from 0.9.1 or below to 1.x</a></li>
<li><a href="#where-to-go-from-here">Where to Go from Here</a></li>
</ul>
<h1 id="overview">Overview</h1>
<p>Spark Streaming is an extension of the core Spark API that enables scalable, high-throughput,
fault-tolerant stream processing of live data streams. Data can be ingested from many sources
like Kafka, Flume, Twitter, ZeroMQ, Kinesis, or TCP sockets, and can be processed using complex
algorithms expressed with high-level functions like <code>map</code>, <code>reduce</code>, <code>join</code> and <code>window</code>.
Finally, processed data can be pushed out to filesystems, databases,
and live dashboards. In fact, you can apply Spark’s
<a href="mllib-guide.html">machine learning</a> and
<a href="graphx-programming-guide.html">graph processing</a> algorithms on data streams.</p>
<p style="text-align: center;">
<img src="img/streaming-arch.png" title="Spark Streaming architecture" alt="Spark Streaming" width="70%" />
</p>
<p>Internally, it works as follows. Spark Streaming receives live input data streams and divides
the data into batches, which are then processed by the Spark engine to generate the final
stream of results in batches.</p>
<p style="text-align: center;">
<img src="img/streaming-flow.png" title="Spark Streaming data flow" alt="Spark Streaming" width="70%" />
</p>
<p>Spark Streaming provides a high-level abstraction called <em>discretized stream</em> or <em>DStream</em>,
which represents a continuous stream of data. DStreams can be created either from input data
streams from sources such as Kafka, Flume, and Kinesis, or by applying high-level
operations on other DStreams. Internally, a DStream is represented as a sequence of
<a href="api/scala/index.html#org.apache.spark.rdd.RDD">RDDs</a>.</p>
<p>This guide shows you how to start writing Spark Streaming programs with DStreams. You can
write Spark Streaming programs in Scala, Java or Python (introduced in Spark 1.2),
all of which are presented in this guide.
You will find tabs throughout this guide that let you choose between code snippets of
different languages.</p>
<p><strong>Note:</strong> There are a few APIs that are either different or not available in Python. Throughout this guide, you will find the tag <span class="badge" style="background-color: grey">Python API</span> highlighting these differences.</p>
<hr />
<h1 id="a-quick-example">A Quick Example</h1>
<p>Before we go into the details of how to write your own Spark Streaming program,
let’s take a quick look at what a simple Spark Streaming program looks like. Let’s say we want to
count the number of words in text data received from a data server listening on a TCP
socket. All you need to
do is as follows.</p>
<div class="codetabs">
<div data-lang="scala">
<p>First, we import the names of the Spark Streaming classes and some implicit
conversions from StreamingContext into our environment in order to add useful methods to
other classes we need (like DStream). <a href="api/scala/index.html#org.apache.spark.streaming.StreamingContext">StreamingContext</a> is the
main entry point for all streaming functionality. We create a local StreamingContext with two execution threads, and a batch interval of 1 second.</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="k">import</span> <span class="nn">org.apache.spark._</span>
<span class="k">import</span> <span class="nn">org.apache.spark.streaming._</span>
<span class="k">import</span> <span class="nn">org.apache.spark.streaming.StreamingContext._</span> <span class="c1">// not necessary since Spark 1.3</span>
<span class="c1">// Create a local StreamingContext with two working thread and batch interval of 1 second.</span>
<span class="c1">// The master requires 2 cores to prevent from a starvation scenario.</span>
<span class="k">val</span> <span class="n">conf</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">SparkConf</span><span class="o">().</span><span class="n">setMaster</span><span class="o">(</span><span class="s">"local[2]"</span><span class="o">).</span><span class="n">setAppName</span><span class="o">(</span><span class="s">"NetworkWordCount"</span><span class="o">)</span>
<span class="k">val</span> <span class="n">ssc</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">StreamingContext</span><span class="o">(</span><span class="n">conf</span><span class="o">,</span> <span class="nc">Seconds</span><span class="o">(</span><span class="mi">1</span><span class="o">))</span></code></pre></div>
<p>Using this context, we can create a DStream that represents streaming data from a TCP
source, specified as hostname (e.g. <code>localhost</code>) and port (e.g. <code>9999</code>).</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="c1">// Create a DStream that will connect to hostname:port, like localhost:9999</span>
<span class="k">val</span> <span class="n">lines</span> <span class="k">=</span> <span class="n">ssc</span><span class="o">.</span><span class="n">socketTextStream</span><span class="o">(</span><span class="s">"localhost"</span><span class="o">,</span> <span class="mi">9999</span><span class="o">)</span></code></pre></div>
<p>This <code>lines</code> DStream represents the stream of data that will be received from the data
server. Each record in this DStream is a line of text. Next, we want to split the lines by
space characters into words.</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="c1">// Split each line into words</span>
<span class="k">val</span> <span class="n">words</span> <span class="k">=</span> <span class="n">lines</span><span class="o">.</span><span class="n">flatMap</span><span class="o">(</span><span class="k">_</span><span class="o">.</span><span class="n">split</span><span class="o">(</span><span class="s">" "</span><span class="o">))</span></code></pre></div>
<p><code>flatMap</code> is a one-to-many DStream operation that creates a new DStream by
generating multiple new records from each record in the source DStream. In this case,
each line will be split into multiple words and the stream of words is represented as the
<code>words</code> DStream. Next, we want to count these words.</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="k">import</span> <span class="nn">org.apache.spark.streaming.StreamingContext._</span> <span class="c1">// not necessary since Spark 1.3</span>
<span class="c1">// Count each word in each batch</span>
<span class="k">val</span> <span class="n">pairs</span> <span class="k">=</span> <span class="n">words</span><span class="o">.</span><span class="n">map</span><span class="o">(</span><span class="n">word</span> <span class="k">=></span> <span class="o">(</span><span class="n">word</span><span class="o">,</span> <span class="mi">1</span><span class="o">))</span>
<span class="k">val</span> <span class="n">wordCounts</span> <span class="k">=</span> <span class="n">pairs</span><span class="o">.</span><span class="n">reduceByKey</span><span class="o">(</span><span class="k">_</span> <span class="o">+</span> <span class="k">_</span><span class="o">)</span>
<span class="c1">// Print the first ten elements of each RDD generated in this DStream to the console</span>
<span class="n">wordCounts</span><span class="o">.</span><span class="n">print</span><span class="o">()</span></code></pre></div>
<p>The <code>words</code> DStream is further mapped (one-to-one transformation) to a DStream of <code>(word,
1)</code> pairs, which is then reduced to get the frequency of words in each batch of data.
Finally, <code>wordCounts.print()</code> will print a few of the counts generated every second.</p>
<p>Note that when these lines are executed, Spark Streaming only sets up the computation it
will perform when it is started, and no real processing has started yet. To start the processing
after all the transformations have been setup, we finally call</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="n">ssc</span><span class="o">.</span><span class="n">start</span><span class="o">()</span> <span class="c1">// Start the computation</span>
<span class="n">ssc</span><span class="o">.</span><span class="n">awaitTermination</span><span class="o">()</span> <span class="c1">// Wait for the computation to terminate</span></code></pre></div>
<p>The complete code can be found in the Spark Streaming example
<a href="https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/NetworkWordCount.scala">NetworkWordCount</a>.
<br /></p>
</div>
<div data-lang="java">
<p>First, we create a
<a href="api/java/index.html?org/apache/spark/streaming/api/java/JavaStreamingContext.html">JavaStreamingContext</a> object,
which is the main entry point for all streaming
functionality. We create a local StreamingContext with two execution threads, and a batch interval of 1 second.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kn">import</span> <span class="nn">org.apache.spark.*</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.apache.spark.api.java.function.*</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.apache.spark.streaming.*</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.apache.spark.streaming.api.java.*</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">scala.Tuple2</span><span class="o">;</span>
<span class="c1">// Create a local StreamingContext with two working thread and batch interval of 1 second</span>
<span class="n">SparkConf</span> <span class="n">conf</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">SparkConf</span><span class="o">().</span><span class="na">setMaster</span><span class="o">(</span><span class="s">"local[2]"</span><span class="o">).</span><span class="na">setAppName</span><span class="o">(</span><span class="s">"NetworkWordCount"</span><span class="o">)</span>
<span class="n">JavaStreamingContext</span> <span class="n">jssc</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">JavaStreamingContext</span><span class="o">(</span><span class="n">conf</span><span class="o">,</span> <span class="n">Durations</span><span class="o">.</span><span class="na">seconds</span><span class="o">(</span><span class="mi">1</span><span class="o">))</span></code></pre></div>
<p>Using this context, we can create a DStream that represents streaming data from a TCP
source, specified as hostname (e.g. <code>localhost</code>) and port (e.g. <code>9999</code>).</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// Create a DStream that will connect to hostname:port, like localhost:9999</span>
<span class="n">JavaReceiverInputDStream</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">lines</span> <span class="o">=</span> <span class="n">jssc</span><span class="o">.</span><span class="na">socketTextStream</span><span class="o">(</span><span class="s">"localhost"</span><span class="o">,</span> <span class="mi">9999</span><span class="o">);</span></code></pre></div>
<p>This <code>lines</code> DStream represents the stream of data that will be received from the data
server. Each record in this stream is a line of text. Then, we want to split the lines by
space into words.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// Split each line into words</span>
<span class="n">JavaDStream</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="n">words</span> <span class="o">=</span> <span class="n">lines</span><span class="o">.</span><span class="na">flatMap</span><span class="o">(</span>
<span class="k">new</span> <span class="n">FlatMapFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">String</span><span class="o">>()</span> <span class="o">{</span>
<span class="nd">@Override</span> <span class="kd">public</span> <span class="n">Iterator</span><span class="o"><</span><span class="n">String</span><span class="o">></span> <span class="nf">call</span><span class="o">(</span><span class="n">String</span> <span class="n">x</span><span class="o">)</span> <span class="o">{</span>
<span class="k">return</span> <span class="n">Arrays</span><span class="o">.</span><span class="na">asList</span><span class="o">(</span><span class="n">x</span><span class="o">.</span><span class="na">split</span><span class="o">(</span><span class="s">" "</span><span class="o">)).</span><span class="na">iterator</span><span class="o">();</span>
<span class="o">}</span>
<span class="o">});</span></code></pre></div>
<p><code>flatMap</code> is a DStream operation that creates a new DStream by
generating multiple new records from each record in the source DStream. In this case,
each line will be split into multiple words and the stream of words is represented as the
<code>words</code> DStream. Note that we defined the transformation using a
<a href="api/scala/index.html#org.apache.spark.api.java.function.FlatMapFunction">FlatMapFunction</a> object.
As we will discover along the way, there are a number of such convenience classes in the Java API
that help define DStream transformations.</p>
<p>Next, we want to count these words.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="c1">// Count each word in each batch</span>
<span class="n">JavaPairDStream</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">></span> <span class="n">pairs</span> <span class="o">=</span> <span class="n">words</span><span class="o">.</span><span class="na">mapToPair</span><span class="o">(</span>
<span class="k">new</span> <span class="n">PairFunction</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>()</span> <span class="o">{</span>
<span class="nd">@Override</span> <span class="kd">public</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">></span> <span class="nf">call</span><span class="o">(</span><span class="n">String</span> <span class="n">s</span><span class="o">)</span> <span class="o">{</span>
<span class="k">return</span> <span class="k">new</span> <span class="n">Tuple2</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>(</span><span class="n">s</span><span class="o">,</span> <span class="mi">1</span><span class="o">);</span>
<span class="o">}</span>
<span class="o">});</span>
<span class="n">JavaPairDStream</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">></span> <span class="n">wordCounts</span> <span class="o">=</span> <span class="n">pairs</span><span class="o">.</span><span class="na">reduceByKey</span><span class="o">(</span>
<span class="k">new</span> <span class="n">Function2</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Integer</span><span class="o">,</span> <span class="n">Integer</span><span class="o">>()</span> <span class="o">{</span>
<span class="nd">@Override</span> <span class="kd">public</span> <span class="n">Integer</span> <span class="nf">call</span><span class="o">(</span><span class="n">Integer</span> <span class="n">i1</span><span class="o">,</span> <span class="n">Integer</span> <span class="n">i2</span><span class="o">)</span> <span class="o">{</span>
<span class="k">return</span> <span class="n">i1</span> <span class="o">+</span> <span class="n">i2</span><span class="o">;</span>
<span class="o">}</span>
<span class="o">});</span>
<span class="c1">// Print the first ten elements of each RDD generated in this DStream to the console</span>
<span class="n">wordCounts</span><span class="o">.</span><span class="na">print</span><span class="o">();</span></code></pre></div>
<p>The <code>words</code> DStream is further mapped (one-to-one transformation) to a DStream of <code>(word,
1)</code> pairs, using a <a href="api/scala/index.html#org.apache.spark.api.java.function.PairFunction">PairFunction</a>
object. Then, it is reduced to get the frequency of words in each batch of data,
using a <a href="api/scala/index.html#org.apache.spark.api.java.function.Function2">Function2</a> object.
Finally, <code>wordCounts.print()</code> will print a few of the counts generated every second.</p>
<p>Note that when these lines are executed, Spark Streaming only sets up the computation it
will perform after it is started, and no real processing has started yet. To start the processing
after all the transformations have been setup, we finally call <code>start</code> method.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">jssc</span><span class="o">.</span><span class="na">start</span><span class="o">();</span> <span class="c1">// Start the computation</span>
<span class="n">jssc</span><span class="o">.</span><span class="na">awaitTermination</span><span class="o">();</span> <span class="c1">// Wait for the computation to terminate</span></code></pre></div>
<p>The complete code can be found in the Spark Streaming example
<a href="https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaNetworkWordCount.java">JavaNetworkWordCount</a>.
<br /></p>
</div>
<div data-lang="python">
<p>First, we import <a href="api/python/pyspark.streaming.html#pyspark.streaming.StreamingContext">StreamingContext</a>, which is the main entry point for all streaming functionality. We create a local StreamingContext with two execution threads, and batch interval of 1 second.</p>
<div class="highlight"><pre><code class="language-python" data-lang="python"><span class="kn">from</span> <span class="nn">pyspark</span> <span class="kn">import</span> <span class="n">SparkContext</span>
<span class="kn">from</span> <span class="nn">pyspark.streaming</span> <span class="kn">import</span> <span class="n">StreamingContext</span>
<span class="c"># Create a local StreamingContext with two working thread and batch interval of 1 second</span>
<span class="n">sc</span> <span class="o">=</span> <span class="n">SparkContext</span><span class="p">(</span><span class="s">"local[2]"</span><span class="p">,</span> <span class="s">"NetworkWordCount"</span><span class="p">)</span>
<span class="n">ssc</span> <span class="o">=</span> <span class="n">StreamingContext</span><span class="p">(</span><span class="n">sc</span><span class="p">,</span> <span class="mi">1</span><span class="p">)</span></code></pre></div>
<p>Using this context, we can create a DStream that represents streaming data from a TCP
source, specified as hostname (e.g. <code>localhost</code>) and port (e.g. <code>9999</code>).</p>
<div class="highlight"><pre><code class="language-python" data-lang="python"><span class="c"># Create a DStream that will connect to hostname:port, like localhost:9999</span>
<span class="n">lines</span> <span class="o">=</span> <span class="n">ssc</span><span class="o">.</span><span class="n">socketTextStream</span><span class="p">(</span><span class="s">"localhost"</span><span class="p">,</span> <span class="mi">9999</span><span class="p">)</span></code></pre></div>
<p>This <code>lines</code> DStream represents the stream of data that will be received from the data
server. Each record in this DStream is a line of text. Next, we want to split the lines by
space into words.</p>
<div class="highlight"><pre><code class="language-python" data-lang="python"><span class="c"># Split each line into words</span>
<span class="n">words</span> <span class="o">=</span> <span class="n">lines</span><span class="o">.</span><span class="n">flatMap</span><span class="p">(</span><span class="k">lambda</span> <span class="n">line</span><span class="p">:</span> <span class="n">line</span><span class="o">.</span><span class="n">split</span><span class="p">(</span><span class="s">" "</span><span class="p">))</span></code></pre></div>
<p><code>flatMap</code> is a one-to-many DStream operation that creates a new DStream by
generating multiple new records from each record in the source DStream. In this case,
each line will be split into multiple words and the stream of words is represented as the
<code>words</code> DStream. Next, we want to count these words.</p>
<div class="highlight"><pre><code class="language-python" data-lang="python"><span class="c"># Count each word in each batch</span>
<span class="n">pairs</span> <span class="o">=</span> <span class="n">words</span><span class="o">.</span><span class="n">map</span><span class="p">(</span><span class="k">lambda</span> <span class="n">word</span><span class="p">:</span> <span class="p">(</span><span class="n">word</span><span class="p">,</span> <span class="mi">1</span><span class="p">))</span>
<span class="n">wordCounts</span> <span class="o">=</span> <span class="n">pairs</span><span class="o">.</span><span class="n">reduceByKey</span><span class="p">(</span><span class="k">lambda</span> <span class="n">x</span><span class="p">,</span> <span class="n">y</span><span class="p">:</span> <span class="n">x</span> <span class="o">+</span> <span class="n">y</span><span class="p">)</span>
<span class="c"># Print the first ten elements of each RDD generated in this DStream to the console</span>
<span class="n">wordCounts</span><span class="o">.</span><span class="n">pprint</span><span class="p">()</span></code></pre></div>
<p>The <code>words</code> DStream is further mapped (one-to-one transformation) to a DStream of <code>(word,
1)</code> pairs, which is then reduced to get the frequency of words in each batch of data.
Finally, <code>wordCounts.pprint()</code> will print a few of the counts generated every second.</p>
<p>Note that when these lines are executed, Spark Streaming only sets up the computation it
will perform when it is started, and no real processing has started yet. To start the processing
after all the transformations have been setup, we finally call</p>
<div class="highlight"><pre><code class="language-python" data-lang="python"><span class="n">ssc</span><span class="o">.</span><span class="n">start</span><span class="p">()</span> <span class="c"># Start the computation</span>
<span class="n">ssc</span><span class="o">.</span><span class="n">awaitTermination</span><span class="p">()</span> <span class="c"># Wait for the computation to terminate</span></code></pre></div>
<p>The complete code can be found in the Spark Streaming example
<a href="https://github.com/apache/spark/blob/master/examples/src/main/python/streaming/network_wordcount.py">NetworkWordCount</a>.
<br /></p>
</div>
</div>
<p>If you have already <a href="index.html#downloading">downloaded</a> and <a href="index.html#building">built</a> Spark,
you can run this example as follows. You will first need to run Netcat
(a small utility found in most Unix-like systems) as a data server by using</p>
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="nv">$ </span>nc -lk 9999</code></pre></div>
<p>Then, in a different terminal, you can start the example by using</p>
<div class="codetabs">
<div data-lang="scala">
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="nv">$ </span>./bin/run-example streaming.NetworkWordCount localhost 9999</code></pre></div>
</div>
<div data-lang="java">
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="nv">$ </span>./bin/run-example streaming.JavaNetworkWordCount localhost 9999</code></pre></div>
</div>
<div data-lang="python">
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="nv">$ </span>./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999</code></pre></div>
</div>
</div>
<p>Then, any lines typed in the terminal running the netcat server will be counted and printed on
screen every second. It will look something like the following.</p>
<table width="100%">
<td>
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="c"># TERMINAL 1:</span>
<span class="c"># Running Netcat</span>
<span class="nv">$ </span>nc -lk 9999
hello world
...</code></pre></div>
</td>
<td width="2%"></td>
<td>
<div class="codetabs">
<div data-lang="scala">
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="c"># TERMINAL 2: RUNNING NetworkWordCount</span>
<span class="nv">$ </span>./bin/run-example streaming.NetworkWordCount localhost 9999
...
-------------------------------------------
Time: <span class="m">1357008430000</span> ms
-------------------------------------------
<span class="o">(</span>hello,1<span class="o">)</span>
<span class="o">(</span>world,1<span class="o">)</span>
...</code></pre></div>
</div>
<div data-lang="java">
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="c"># TERMINAL 2: RUNNING JavaNetworkWordCount</span>
<span class="nv">$ </span>./bin/run-example streaming.JavaNetworkWordCount localhost 9999
...
-------------------------------------------
Time: <span class="m">1357008430000</span> ms
-------------------------------------------
<span class="o">(</span>hello,1<span class="o">)</span>
<span class="o">(</span>world,1<span class="o">)</span>
...</code></pre></div>
</div>
<div data-lang="python">
<div class="highlight"><pre><code class="language-bash" data-lang="bash"><span class="c"># TERMINAL 2: RUNNING network_wordcount.py</span>
<span class="nv">$ </span>./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999
...
-------------------------------------------
Time: 2014-10-14 15:25:21
-------------------------------------------
<span class="o">(</span>hello,1<span class="o">)</span>
<span class="o">(</span>world,1<span class="o">)</span>
...</code></pre></div>
</div>
</div>
</td>
</table>
<hr />
<hr />
<h1 id="basic-concepts">Basic Concepts</h1>
<p>Next, we move beyond the simple example and elaborate on the basics of Spark Streaming.</p>
<h2 id="linking">Linking</h2>
<p>Similar to Spark, Spark Streaming is available through Maven Central. To write your own Spark Streaming program, you will have to add the following dependency to your SBT or Maven project.</p>
<div class="codetabs">
<div data-lang="Maven">
<pre><code><dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.0.0-SNAPSHOT</version>
</dependency>
</code></pre>
</div>
<div data-lang="SBT">
<pre><code>libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "2.0.0-SNAPSHOT"
</code></pre>
</div>
</div>
<p>For ingesting data from sources like Kafka, Flume, and Kinesis that are not present in the Spark
Streaming core
API, you will have to add the corresponding
artifact <code>spark-streaming-xyz_2.11</code> to the dependencies. For example,
some of the common ones are as follows.</p>
<table class="table">
<tr><th>Source</th><th>Artifact</th></tr>
<tr><td> Kafka </td><td> spark-streaming-kafka_2.11 </td></tr>
<tr><td> Flume </td><td> spark-streaming-flume_2.11 </td></tr>
<tr><td> Kinesis<br /></td><td>spark-streaming-kinesis-asl_2.11 [Amazon Software License] </td></tr>
<tr><td> Twitter </td><td> spark-streaming-twitter_2.11 </td></tr>
<tr><td> ZeroMQ </td><td> spark-streaming-zeromq_2.11 </td></tr>
<tr><td> MQTT </td><td> spark-streaming-mqtt_2.11 </td></tr>
<tr><td></td><td></td></tr>
</table>
<p>For an up-to-date list, please refer to the
<a href="http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.apache.spark%22%20AND%20v%3A%222.0.0%22">Maven repository</a>
for the full list of supported sources and artifacts.</p>
<hr />
<h2 id="initializing-streamingcontext">Initializing StreamingContext</h2>
<p>To initialize a Spark Streaming program, a <strong>StreamingContext</strong> object has to be created which is the main entry point of all Spark Streaming functionality.</p>
<div class="codetabs">
<div data-lang="scala">
<p>A <a href="api/scala/index.html#org.apache.spark.streaming.StreamingContext">StreamingContext</a> object can be created from a <a href="api/scala/index.html#org.apache.spark.SparkConf">SparkConf</a> object.</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="k">import</span> <span class="nn">org.apache.spark._</span>
<span class="k">import</span> <span class="nn">org.apache.spark.streaming._</span>
<span class="k">val</span> <span class="n">conf</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">SparkConf</span><span class="o">().</span><span class="n">setAppName</span><span class="o">(</span><span class="n">appName</span><span class="o">).</span><span class="n">setMaster</span><span class="o">(</span><span class="n">master</span><span class="o">)</span>
<span class="k">val</span> <span class="n">ssc</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">StreamingContext</span><span class="o">(</span><span class="n">conf</span><span class="o">,</span> <span class="nc">Seconds</span><span class="o">(</span><span class="mi">1</span><span class="o">))</span></code></pre></div>
<p>The <code>appName</code> parameter is a name for your application to show on the cluster UI.
<code>master</code> is a <a href="submitting-applications.html#master-urls">Spark, Mesos or YARN cluster URL</a>,
or a special <strong>“local[*]”</strong> string to run in local mode. In practice, when running on a cluster,
you will not want to hardcode <code>master</code> in the program,
but rather <a href="submitting-applications.html">launch the application with <code>spark-submit</code></a> and
receive it there. However, for local testing and unit tests, you can pass “local[*]” to run Spark Streaming
in-process (detects the number of cores in the local system). Note that this internally creates a <a href="api/scala/index.html#org.apache.spark.SparkContext">SparkContext</a> (starting point of all Spark functionality) which can be accessed as <code>ssc.sparkContext</code>.</p>
<p>The batch interval must be set based on the latency requirements of your application
and available cluster resources. See the <a href="#setting-the-right-batch-interval">Performance Tuning</a>
section for more details.</p>
<p>A <code>StreamingContext</code> object can also be created from an existing <code>SparkContext</code> object.</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="k">import</span> <span class="nn">org.apache.spark.streaming._</span>
<span class="k">val</span> <span class="n">sc</span> <span class="k">=</span> <span class="o">...</span> <span class="c1">// existing SparkContext</span>
<span class="k">val</span> <span class="n">ssc</span> <span class="k">=</span> <span class="k">new</span> <span class="nc">StreamingContext</span><span class="o">(</span><span class="n">sc</span><span class="o">,</span> <span class="nc">Seconds</span><span class="o">(</span><span class="mi">1</span><span class="o">))</span></code></pre></div>
</div>
<div data-lang="java">
<p>A <a href="api/java/index.html?org/apache/spark/streaming/api/java/JavaStreamingContext.html">JavaStreamingContext</a> object can be created from a <a href="api/java/index.html?org/apache/spark/SparkConf.html">SparkConf</a> object.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kn">import</span> <span class="nn">org.apache.spark.*</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.apache.spark.streaming.api.java.*</span><span class="o">;</span>
<span class="n">SparkConf</span> <span class="n">conf</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">SparkConf</span><span class="o">().</span><span class="na">setAppName</span><span class="o">(</span><span class="n">appName</span><span class="o">).</span><span class="na">setMaster</span><span class="o">(</span><span class="n">master</span><span class="o">);</span>
<span class="n">JavaStreamingContext</span> <span class="n">ssc</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">JavaStreamingContext</span><span class="o">(</span><span class="n">conf</span><span class="o">,</span> <span class="n">Duration</span><span class="o">(</span><span class="mi">1000</span><span class="o">));</span></code></pre></div>
<p>The <code>appName</code> parameter is a name for your application to show on the cluster UI.
<code>master</code> is a <a href="submitting-applications.html#master-urls">Spark, Mesos or YARN cluster URL</a>,
or a special <strong>“local[*]”</strong> string to run in local mode. In practice, when running on a cluster,
you will not want to hardcode <code>master</code> in the program,
but rather <a href="submitting-applications.html">launch the application with <code>spark-submit</code></a> and
receive it there. However, for local testing and unit tests, you can pass “local[*]” to run Spark Streaming
in-process. Note that this internally creates a <a href="api/java/index.html?org/apache/spark/api/java/JavaSparkContext.html">JavaSparkContext</a> (starting point of all Spark functionality) which can be accessed as <code>ssc.sparkContext</code>.</p>
<p>The batch interval must be set based on the latency requirements of your application
and available cluster resources. See the <a href="#setting-the-right-batch-interval">Performance Tuning</a>
section for more details.</p>
<p>A <code>JavaStreamingContext</code> object can also be created from an existing <code>JavaSparkContext</code>.</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kn">import</span> <span class="nn">org.apache.spark.streaming.api.java.*</span><span class="o">;</span>
<span class="n">JavaSparkContext</span> <span class="n">sc</span> <span class="o">=</span> <span class="o">...</span> <span class="c1">//existing JavaSparkContext</span>
<span class="n">JavaStreamingContext</span> <span class="n">ssc</span> <span class="o">=</span> <span class="k">new</span> <span class="nf">JavaStreamingContext</span><span class="o">(</span><span class="n">sc</span><span class="o">,</span> <span class="n">Durations</span><span class="o">.</span><span class="na">seconds</span><span class="o">(</span><span class="mi">1</span><span class="o">));</span></code></pre></div>
</div>
<div data-lang="python">
<p>A <a href="api/python/pyspark.streaming.html#pyspark.streaming.StreamingContext">StreamingContext</a> object can be created from a <a href="api/python/pyspark.html#pyspark.SparkContext">SparkContext</a> object.</p>
<div class="highlight"><pre><code class="language-python" data-lang="python"><span class="kn">from</span> <span class="nn">pyspark</span> <span class="kn">import</span> <span class="n">SparkContext</span>
<span class="kn">from</span> <span class="nn">pyspark.streaming</span> <span class="kn">import</span> <span class="n">StreamingContext</span>
<span class="n">sc</span> <span class="o">=</span> <span class="n">SparkContext</span><span class="p">(</span><span class="n">master</span><span class="p">,</span> <span class="n">appName</span><span class="p">)</span>
<span class="n">ssc</span> <span class="o">=</span> <span class="n">StreamingContext</span><span class="p">(</span><span class="n">sc</span><span class="p">,</span> <span class="mi">1</span><span class="p">)</span></code></pre></div>
<p>The <code>appName</code> parameter is a name for your application to show on the cluster UI.
<code>master</code> is a <a href="submitting-applications.html#master-urls">Spark, Mesos or YARN cluster URL</a>,
or a special <strong>“local[*]”</strong> string to run in local mode. In practice, when running on a cluster,
you will not want to hardcode <code>master</code> in the program,
but rather <a href="submitting-applications.html">launch the application with <code>spark-submit</code></a> and
receive it there. However, for local testing and unit tests, you can pass “local[*]” to run Spark Streaming
in-process (detects the number of cores in the local system).</p>
<p>The batch interval must be set based on the latency requirements of your application
and available cluster resources. See the <a href="#setting-the-right-batch-interval">Performance Tuning</a>
section for more details.</p>
</div>
</div>
<p>After a context is defined, you have to do the following.</p>
<ol>
<li>Define the input sources by creating input DStreams.</li>
<li>Define the streaming computations by applying transformation and output operations to DStreams.</li>
<li>Start receiving data and processing it using <code>streamingContext.start()</code>.</li>
<li>Wait for the processing to be stopped (manually or due to any error) using <code>streamingContext.awaitTermination()</code>.</li>
<li>The processing can be manually stopped using <code>streamingContext.stop()</code>.</li>
</ol>
<h5 class="no_toc" id="points-to-remember">Points to remember:</h5>
<ul>
<li>Once a context has been started, no new streaming computations can be set up or added to it.</li>
<li>Once a context has been stopped, it cannot be restarted.</li>
<li>Only one StreamingContext can be active in a JVM at the same time.</li>
<li>stop() on StreamingContext also stops the SparkContext. To stop only the StreamingContext, set the optional parameter of <code>stop()</code> called <code>stopSparkContext</code> to false.</li>
<li>A SparkContext can be re-used to create multiple StreamingContexts, as long as the previous StreamingContext is stopped (without stopping the SparkContext) before the next StreamingContext is created.</li>
</ul>
<hr />
<h2 id="discretized-streams-dstreams">Discretized Streams (DStreams)</h2>
<p><strong>Discretized Stream</strong> or <strong>DStream</strong> is the basic abstraction provided by Spark Streaming.
It represents a continuous stream of data, either the input data stream received from source,
or the processed data stream generated by transforming the input stream. Internally,
a DStream is represented by a continuous series of RDDs, which is Spark’s abstraction of an immutable,
distributed dataset (see <a href="programming-guide.html#resilient-distributed-datasets-rdds">Spark Programming Guide</a> for more details). Each RDD in a DStream contains data from a certain interval,
as shown in the following figure.</p>
<p style="text-align: center;">
<img src="img/streaming-dstream.png" title="Spark Streaming data flow" alt="Spark Streaming" width="70%" />
</p>
<p>Any operation applied on a DStream translates to operations on the underlying RDDs. For example,
in the <a href="#a-quick-example">earlier example</a> of converting a stream of lines to words,
the <code>flatMap</code> operation is applied on each RDD in the <code>lines</code> DStream to generate the RDDs of the
<code>words</code> DStream. This is shown in the following figure.</p>
<p style="text-align: center;">
<img src="img/streaming-dstream-ops.png" title="Spark Streaming data flow" alt="Spark Streaming" width="70%" />
</p>
<p>These underlying RDD transformations are computed by the Spark engine. The DStream operations
hide most of these details and provide the developer with a higher-level API for convenience.
These operations are discussed in detail in later sections.</p>
<hr />
<h2 id="input-dstreams-and-receivers">Input DStreams and Receivers</h2>
<p>Input DStreams are DStreams representing the stream of input data received from streaming
sources. In the <a href="#a-quick-example">quick example</a>, <code>lines</code> was an input DStream as it represented
the stream of data received from the netcat server. Every input DStream
(except file stream, discussed later in this section) is associated with a <strong>Receiver</strong>
(<a href="api/scala/index.html#org.apache.spark.streaming.receiver.Receiver">Scala doc</a>,
<a href="api/java/org/apache/spark/streaming/receiver/Receiver.html">Java doc</a>) object which receives the
data from a source and stores it in Spark’s memory for processing.</p>
<p>Spark Streaming provides two categories of built-in streaming sources.</p>
<ul>
<li><em>Basic sources</em>: Sources directly available in the StreamingContext API.
Examples: file systems, socket connections, and Akka actors.</li>
<li><em>Advanced sources</em>: Sources like Kafka, Flume, Kinesis, Twitter, etc. are available through
extra utility classes. These require linking against extra dependencies as discussed in the
<a href="#linking">linking</a> section.</li>
</ul>
<p>We are going to discuss some of the sources present in each category later in this section.</p>
<p>Note that, if you want to receive multiple streams of data in parallel in your streaming
application, you can create multiple input DStreams (discussed
further in the <a href="#level-of-parallelism-in-data-receiving">Performance Tuning</a> section). This will
create multiple receivers which will simultaneously receive multiple data streams. But note that a
Spark worker/executor is a long-running task, hence it occupies one of the cores allocated to the
Spark Streaming application. Therefore, it is important to remember that a Spark Streaming application
needs to be allocated enough cores (or threads, if running locally) to process the received data,
as well as to run the receiver(s).</p>
<h5 class="no_toc" id="points-to-remember-1">Points to remember</h5>
<ul>
<li>
<p>When running a Spark Streaming program locally, do not use “local” or “local[1]” as the master URL.
Either of these means that only one thread will be used for running tasks locally. If you are using
a input DStream based on a receiver (e.g. sockets, Kafka, Flume, etc.), then the single thread will
be used to run the receiver, leaving no thread for processing the received data. Hence, when
running locally, always use “local[<em>n</em>]” as the master URL, where <em>n</em> > number of receivers to run
(see <a href="configuration.html#spark-properties">Spark Properties</a> for information on how to set
the master).</p>
</li>
<li>
<p>Extending the logic to running on a cluster, the number of cores allocated to the Spark Streaming
application must be more than the number of receivers. Otherwise the system will receive data, but
not be able to process it.</p>
</li>
</ul>
<h3 class="no_toc" id="basic-sources">Basic Sources</h3>
<p>We have already taken a look at the <code>ssc.socketTextStream(...)</code> in the <a href="#a-quick-example">quick example</a>
which creates a DStream from text
data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
methods for creating DStreams from files and Akka actors as input sources.</p>
<ul>
<li>
<p><strong>File Streams:</strong> For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:</p>
<div class="codetabs">
<div data-lang="scala">
<pre><code> streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
</code></pre>
</div>
<div data-lang="java">
<pre><code> streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
</code></pre>
</div>
<div data-lang="python">
<pre><code> streamingContext.textFileStream(dataDirectory)
</code></pre>
</div>
</div>
<p>Spark Streaming will monitor the directory <code>dataDirectory</code> and process any files created in that directory (files written in nested directories not supported). Note that</p>
<ul>
<li>The files must have the same data format.</li>
<li>The files must be created in the <code>dataDirectory</code> by atomically <em>moving</em> or <em>renaming</em> them into
the data directory.</li>
<li>Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.</li>
</ul>
<p>For simple text files, there is an easier method <code>streamingContext.textFileStream(dataDirectory)</code>. And file streams do not require running a receiver, hence does not require allocating cores.</p>
<p><span class="badge" style="background-color: grey">Python API</span> <code>fileStream</code> is not available in the Python API, only <code>textFileStream</code> is available.</p>
</li>
<li>
<p><strong>Streams based on Custom Actors:</strong> DStreams can be created with data streams received through Akka
actors by using <code>AkkaUtils.createStream(ssc, actorProps, actor-name)</code>. See the <a href="streaming-custom-receivers.html">Custom Receiver
Guide</a> for more details.</p>
<p><span class="badge" style="background-color: grey">Python API</span> Since actors are available only in the Java and Scala
libraries, <code>AkkaUtils.createStream</code> is not available in the Python API.</p>
</li>
<li>
<p><strong>Queue of RDDs as a Stream:</strong> For testing a Spark Streaming application with test data, one can also create a DStream based on a queue of RDDs, using <code>streamingContext.queueStream(queueOfRDDs)</code>. Each RDD pushed into the queue will be treated as a batch of data in the DStream, and processed like a stream.</p>
</li>
</ul>
<p>For more details on streams from sockets, files, and actors,
see the API documentations of the relevant functions in
<a href="api/scala/index.html#org.apache.spark.streaming.StreamingContext">StreamingContext</a> for
Scala, <a href="api/java/index.html?org/apache/spark/streaming/api/java/JavaStreamingContext.html">JavaStreamingContext</a>
for Java, and <a href="api/python/pyspark.streaming.html#pyspark.streaming.StreamingContext">StreamingContext</a> for Python.</p>
<h3 class="no_toc" id="advanced-sources">Advanced Sources</h3>
<p><span class="badge" style="background-color: grey">Python API</span> As of Spark 2.0.0,
out of these sources, Kafka, Kinesis, Flume and MQTT are available in the Python API.</p>
<p>This category of sources require interfacing with external non-Spark libraries, some of them with
complex dependencies (e.g., Kafka and Flume). Hence, to minimize issues related to version conflicts
of dependencies, the functionality to create DStreams from these sources has been moved to separate
libraries that can be <a href="#linking">linked</a> to explicitly when necessary. For example, if you want to
create a DStream using data from Twitter’s stream of tweets, you have to do the following:</p>
<ol>
<li><em>Linking</em>: Add the artifact <code>spark-streaming-twitter_2.11</code> to the
SBT/Maven project dependencies.</li>
<li><em>Programming</em>: Import the <code>TwitterUtils</code> class and create a DStream with
<code>TwitterUtils.createStream</code> as shown below.</li>
<li><em>Deploying</em>: Generate an uber JAR with all the dependencies (including the dependency
<code>spark-streaming-twitter_2.11</code> and its transitive dependencies) and
then deploy the application. This is further explained in the <a href="#deploying-applications">Deploying section</a>.</li>
</ol>
<div class="codetabs">
<div data-lang="scala">
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="k">import</span> <span class="nn">org.apache.spark.streaming.twitter._</span>
<span class="nc">TwitterUtils</span><span class="o">.</span><span class="n">createStream</span><span class="o">(</span><span class="n">ssc</span><span class="o">,</span> <span class="nc">None</span><span class="o">)</span></code></pre></div>
</div>
<div data-lang="java">
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="kn">import</span> <span class="nn">org.apache.spark.streaming.twitter.*</span><span class="o">;</span>
<span class="n">TwitterUtils</span><span class="o">.</span><span class="na">createStream</span><span class="o">(</span><span class="n">jssc</span><span class="o">);</span></code></pre></div>
</div>
</div>
<p>Note that these advanced sources are not available in the Spark shell, hence applications based on
these advanced sources cannot be tested in the shell. If you really want to use them in the Spark
shell you will have to download the corresponding Maven artifact’s JAR along with its dependencies
and add it to the classpath.</p>
<p>Some of these advanced sources are as follows.</p>
<ul>
<li>
<p><strong>Kafka:</strong> Spark Streaming 2.0.0 is compatible with Kafka 0.8.2.1. See the <a href="streaming-kafka-integration.html">Kafka Integration Guide</a> for more details.</p>
</li>
<li>
<p><strong>Flume:</strong> Spark Streaming 2.0.0 is compatible with Flume 1.6.0. See the <a href="streaming-flume-integration.html">Flume Integration Guide</a> for more details.</p>
</li>
<li>
<p><strong>Kinesis:</strong> Spark Streaming 2.0.0 is compatible with Kinesis Client Library 1.2.1. See the <a href="streaming-kinesis-integration.html">Kinesis Integration Guide</a> for more details.</p>
</li>
<li>
<p><strong>Twitter:</strong> Spark Streaming’s TwitterUtils uses Twitter4j to get the public stream of tweets using
<a href="https://dev.twitter.com/docs/streaming-apis">Twitter’s Streaming API</a>. Authentication information
can be provided by any of the <a href="http://twitter4j.org/en/configuration.html">methods</a> supported by
Twitter4J library. You can either get the public stream, or get the filtered stream based on a
keywords. See the API documentation (<a href="api/scala/index.html#org.apache.spark.streaming.twitter.TwitterUtils$">Scala</a>,
<a href="api/java/index.html?org/apache/spark/streaming/twitter/TwitterUtils.html">Java</a>) and examples
(<a href="https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/TwitterPopularTags.scala">TwitterPopularTags</a>
and <a href="https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/TwitterAlgebirdCMS.scala">TwitterAlgebirdCMS</a>).</p>
</li>
</ul>
<h3 class="no_toc" id="custom-sources">Custom Sources</h3>
<p><span class="badge" style="background-color: grey">Python API</span> This is not yet supported in Python.</p>
<p>Input DStreams can also be created out of custom data sources. All you have to do is implement a
user-defined <strong>receiver</strong> (see next section to understand what that is) that can receive data from
the custom sources and push it into Spark. See the <a href="streaming-custom-receivers.html">Custom Receiver
Guide</a> for details.</p>
<h3 class="no_toc" id="receiver-reliability">Receiver Reliability</h3>
<p>There can be two kinds of data sources based on their <em>reliability</em>. Sources
(like Kafka and Flume) allow the transferred data to be acknowledged. If the system receiving
data from these <em>reliable</em> sources acknowledges the received data correctly, it can be ensured
that no data will be lost due to any kind of failure. This leads to two kinds of receivers:</p>
<ol>
<li><em>Reliable Receiver</em> - A <em>reliable receiver</em> correctly sends acknowledgment to a reliable
source when the data has been received and stored in Spark with replication.</li>
<li><em>Unreliable Receiver</em> - An <em>unreliable receiver</em> does <em>not</em> send acknowledgment to a source. This can be used for sources that do not support acknowledgment, or even for reliable sources when one does not want or need to go into the complexity of acknowledgment.</li>
</ol>
<p>The details of how to write a reliable receiver are discussed in the
<a href="streaming-custom-receivers.html">Custom Receiver Guide</a>.</p>
<hr />
<h2 id="transformations-on-dstreams">Transformations on DStreams</h2>
<p>Similar to that of RDDs, transformations allow the data from the input DStream to be modified.
DStreams support many of the transformations available on normal Spark RDD’s.
Some of the common ones are as follows.</p>
<table class="table">
<tr><th style="width:25%">Transformation</th><th>Meaning</th></tr>
<tr>
<td> <b>map</b>(<i>func</i>) </td>
<td> Return a new DStream by passing each element of the source DStream through a
function <i>func</i>. </td>
</tr>
<tr>
<td> <b>flatMap</b>(<i>func</i>) </td>
<td> Similar to map, but each input item can be mapped to 0 or more output items. </td>
</tr>
<tr>
<td> <b>filter</b>(<i>func</i>) </td>
<td> Return a new DStream by selecting only the records of the source DStream on which
<i>func</i> returns true. </td>
</tr>
<tr>
<td> <b>repartition</b>(<i>numPartitions</i>) </td>
<td> Changes the level of parallelism in this DStream by creating more or fewer partitions. </td>
</tr>
<tr>
<td> <b>union</b>(<i>otherStream</i>) </td>
<td> Return a new DStream that contains the union of the elements in the source DStream and
<i>otherDStream</i>. </td>
</tr>
<tr>
<td> <b>count</b>() </td>
<td> Return a new DStream of single-element RDDs by counting the number of elements in each RDD
of the source DStream. </td>
</tr>
<tr>
<td> <b>reduce</b>(<i>func</i>) </td>
<td> Return a new DStream of single-element RDDs by aggregating the elements in each RDD of the
source DStream using a function <i>func</i> (which takes two arguments and returns one).
The function should be associative and commutative so that it can be computed in parallel. </td>
</tr>
<tr>
<td> <b>countByValue</b>() </td>
<td> When called on a DStream of elements of type K, return a new DStream of (K, Long) pairs
where the value of each key is its frequency in each RDD of the source DStream. </td>
</tr>
<tr>
<td> <b>reduceByKey</b>(<i>func</i>, [<i>numTasks</i>]) </td>
<td> When called on a DStream of (K, V) pairs, return a new DStream of (K, V) pairs where the
values for each key are aggregated using the given reduce function. <b>Note:</b> By default,
this uses Spark's default number of parallel tasks (2 for local mode, and in cluster mode the number
is determined by the config property <code>spark.default.parallelism</code>) to do the grouping.
You can pass an optional <code>numTasks</code> argument to set a different number of tasks.</td>
</tr>
<tr>
<td> <b>join</b>(<i>otherStream</i>, [<i>numTasks</i>]) </td>
<td> When called on two DStreams of (K, V) and (K, W) pairs, return a new DStream of (K, (V, W))
pairs with all pairs of elements for each key. </td>
</tr>
<tr>
<td> <b>cogroup</b>(<i>otherStream</i>, [<i>numTasks</i>]) </td>
<td> When called on a DStream of (K, V) and (K, W) pairs, return a new DStream of
(K, Seq[V], Seq[W]) tuples.</td>
</tr>
<tr>
<td> <b>transform</b>(<i>func</i>) </td>
<td> Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream.
This can be used to do arbitrary RDD operations on the DStream. </td>
</tr>
<tr>
<td> <b>updateStateByKey</b>(<i>func</i>) </td>
<td> Return a new "state" DStream where the state for each key is updated by applying the
given function on the previous state of the key and the new values for the key. This can be
used to maintain arbitrary state data for each key.</td>
</tr>
<tr><td></td><td></td></tr>
</table>
<p>A few of these transformations are worth discussing in more detail.</p>
<h4 class="no_toc" id="updatestatebykey-operation">UpdateStateByKey Operation</h4>
<p>The <code>updateStateByKey</code> operation allows you to maintain arbitrary state while continuously updating
it with new information. To use this, you will have to do two steps.</p>
<ol>
<li>Define the state - The state can be an arbitrary data type.</li>
<li>Define the state update function - Specify with a function how to update the state using the
previous state and the new values from an input stream.</li>
</ol>
<p>In every batch, Spark will apply the state update function for all existing keys, regardless of whether they have new data in a batch or not. If the update function returns <code>None</code> then the key-value pair will be eliminated.</p>
<p>Let’s illustrate this with an example. Say you want to maintain a running count of each word
seen in a text data stream. Here, the running count is the state and it is an integer. We
define the update function as:</p>
<div class="codetabs">
<div data-lang="scala">
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="k">def</span> <span class="n">updateFunction</span><span class="o">(</span><span class="n">newValues</span><span class="k">:</span> <span class="kt">Seq</span><span class="o">[</span><span class="kt">Int</span><span class="o">],</span> <span class="n">runningCount</span><span class="k">:</span> <span class="kt">Option</span><span class="o">[</span><span class="kt">Int</span><span class="o">])</span><span class="k">:</span> <span class="kt">Option</span><span class="o">[</span><span class="kt">Int</span><span class="o">]</span> <span class="k">=</span> <span class="o">{</span>
<span class="k">val</span> <span class="n">newCount</span> <span class="k">=</span> <span class="o">...</span> <span class="c1">// add the new values with the previous running count to get the new count</span>
<span class="nc">Some</span><span class="o">(</span><span class="n">newCount</span><span class="o">)</span>
<span class="o">}</span></code></pre></div>
<p>This is applied on a DStream containing words (say, the <code>pairs</code> DStream containing <code>(word,
1)</code> pairs in the <a href="#a-quick-example">earlier example</a>).</p>
<div class="highlight"><pre><code class="language-scala" data-lang="scala"><span class="k">val</span> <span class="n">runningCounts</span> <span class="k">=</span> <span class="n">pairs</span><span class="o">.</span><span class="n">updateStateByKey</span><span class="o">[</span><span class="kt">Int</span><span class="o">](</span><span class="n">updateFunction</span> <span class="k">_</span><span class="o">)</span></code></pre></div>
<p>The update function will be called for each word, with <code>newValues</code> having a sequence of 1’s (from
the <code>(word, 1)</code> pairs) and the <code>runningCount</code> having the previous count. For the complete
Scala code, take a look at the example
<a href="https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache
/spark/examples/streaming/StatefulNetworkWordCount.scala">StatefulNetworkWordCount.scala</a>.</p>
</div>
<div data-lang="java">
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">Function2</span><span class="o"><</span><span class="n">List</span><span class="o"><</span><span class="n">Integer</span><span class="o">>,</span> <span class="n">Optional</span><span class="o"><</span><span class="n">Integer</span><span class="o">>,</span> <span class="n">Optional</span><span class="o"><</span><span class="n">Integer</span><span class="o">>></span> <span class="n">updateFunction</span> <span class="o">=</span>
<span class="k">new</span> <span class="n">Function2</span><span class="o"><</span><span class="n">List</span><span class="o"><</span><span class="n">Integer</span><span class="o">>,</span> <span class="n">Optional</span><span class="o"><</span><span class="n">Integer</span><span class="o">>,</span> <span class="n">Optional</span><span class="o"><</span><span class="n">Integer</span><span class="o">>>()</span> <span class="o">{</span>
<span class="nd">@Override</span> <span class="kd">public</span> <span class="n">Optional</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="nf">call</span><span class="o">(</span><span class="n">List</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">values</span><span class="o">,</span> <span class="n">Optional</span><span class="o"><</span><span class="n">Integer</span><span class="o">></span> <span class="n">state</span><span class="o">)</span> <span class="o">{</span>
<span class="n">Integer</span> <span class="n">newSum</span> <span class="o">=</span> <span class="o">...</span> <span class="c1">// add the new values with the previous running count to get the new count</span>
<span class="k">return</span> <span class="n">Optional</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="n">newSum</span><span class="o">);</span>
<span class="o">}</span>
<span class="o">};</span></code></pre></div>
<p>This is applied on a DStream containing words (say, the <code>pairs</code> DStream containing <code>(word,
1)</code> pairs in the <a href="#a-quick-example">quick example</a>).</p>
<div class="highlight"><pre><code class="language-java" data-lang="java"><span class="n">JavaPairDStream</span><span class="o"><</span><span class="n">String</span><span class="o">,</span> <span class="n">Integer</span><span class="o">></span> <span class="n">runningCounts</span> <span class="o">=</span> <span class="n">pairs</span><span class="o">.</span><span class="na">updateStateByKey</span><span class="o">(</span><span class="n">updateFunction</span><span class="o">);</span></code></pre></div>
<p>The update function will be called for each word, with <code>newValues</code> having a sequence of 1’s (from
the <code>(word, 1)</code> pairs) and the <code>runningCount</code> having the previous count. For the complete