第10讲：Flink Side OutPut 分流

admin

9 h: ?( H! p  [& v<h4 id="flink系列文章">Flink系列文章</h4>
  D' M, [2 {: ?9 J7 v3 }7 m3 R<ol>
<li><a  href="https://www.ikeguang.com/?p=1976">第01讲：Flink 的应用场景和架构模型</a></li>
<li><a  href="https://www.ikeguang.com/?p=1977">第02讲：Flink 入门程序 WordCount 和 SQL 实现</a></li>
( h3 X8 _7 U; y<li><a  href="https://www.ikeguang.com/?p=1978">第03讲：Flink 的编程模型与其他框架比较</a></li>
<li><a  href="https://www.ikeguang.com/?p=1982">第04讲：Flink 常用的 DataSet 和 DataStream API</a></li>
9 x" r/ n' J  G0 f<li><a  href="https://www.ikeguang.com/?p=1983">第05讲：Flink SQL &amp; Table 编程和案例</a></li>
& e: N. X5 M5 Z& R1 S<li><a  href="https://www.ikeguang.com/?p=1985">第06讲：Flink 集群安装部署和 HA 配置</a></li>
2 W( K. ~7 H5 D<li><a  href="https://www.ikeguang.com/?p=1986">第07讲：Flink 常见核心概念分析</a></li>
- I* H! R. `2 ~# A' S2 T<li><a  href="https://www.ikeguang.com/?p=1987">第08讲：Flink 窗口、时间和水印</a></li>
<li><a  href="https://www.ikeguang.com/?p=1988">第09讲：Flink 状态与容错</a></li>
</ol>
  x0 [/ J6 H) m, n' k$ M! b<blockquote>
<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
</blockquote>
# A% g# `5 Q, V4 v9 E/ \+ N7 x<p>这一课时将介绍 Flink 中提供的一个很重要的功能：旁路分流器。</p>
<h3 id="分流场景">分流场景</h3>
<p>我们在生产实践中经常会遇到这样的场景，需把输入源按照需要进行拆分，比如我期望把订单流按照金额大小进行拆分，或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢？</p>
; B, j0 ], V* \& O2 u7 r<h3 id="分流的方法">分流的方法</h3>
) [6 z" w+ x! ?8 a: C0 h<p>通常来说针对不同的场景，有以下三种办法进行流的拆分。</p>
<h4 id="filter-分流">Filter 分流</h4>
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CAy6ADUaXAACSFUbdpuA911-20210223084827182.png" ></p>
<p>Filter 方法我们在第 04 课时中（Flink 常用的 DataSet 和 DataStream API）讲过，这个算子用来根据用户输入的条件进行过滤，每个元素都会被 filter() 函数处理，如果 filter() 函数返回 true 则保留，否则丢弃。那么用在分流的场景，我们可以做多次 filter，把我们需要的不同数据生成不同的流。</p>
8 J! f0 d' t! X# K, U3 I* w+ s<p>来看下面的例子：</p>
- Q' d" v5 a3 u  E: {1 a( i<p>复制代码</p>
  H9 N6 ?) X$ `$ W7 D<pre><code class="language-java">public static void main(String[] args) throws Exception {
' M. |0 J! Z% R: T    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
$ w& G$ b4 G( N# S) _9 l, N0 K    //获取数据源
    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
    data.add(new Tuple3&lt;&gt;(0,1,0));
    data.add(new Tuple3&lt;&gt;(0,1,1));
    data.add(new Tuple3&lt;&gt;(0,2,2));
+ x, s' s; n# k! A- `5 e$ f1 O    data.add(new Tuple3&lt;&gt;(0,1,3));
    data.add(new Tuple3&lt;&gt;(1,2,5));
7 Q- Z& Z5 J2 \. i! l. F$ w    data.add(new Tuple3&lt;&gt;(1,2,9));
- Y7 _! b+ g% |5 J$ |& b& Q    data.add(new Tuple3&lt;&gt;(1,2,11));
    data.add(new Tuple3&lt;&gt;(1,2,13));
5 t+ ~# y4 q( e2 J& E: g" O
" C! q0 t* c" _& f1 M    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
8 B. ?9 Z  X) N9 g: J' o
# E# y0 w/ n, J& G+ ^9 Z3 H( E9 }4 Q7 O
' _3 l/ I# F2 a& ~1 g
( T* a$ A) N$ s3 \8 g  t    SingleOutputStreamOperator&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; zeroStream = items.filter((FilterFunction&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt;) value -&gt; value.f0 == 0);

    SingleOutputStreamOperator&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; oneStream = items.filter((FilterFunction&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt;) value -&gt; value.f0 == 1);

3 {! B) y! T8 U! q6 [
6 [  p' M6 P8 E* f  F! \! E5 _
) |! w6 \# G" d: C9 T    zeroStream.print();

    oneStream.printToErr();
. ^3 z2 ?- s8 t) i

$ F2 B% m- e2 G3 \- w" y& }
) c' i* s/ h% ^8 c. x0 j: f
. D+ G" h: E# n1 |0 U4 t" D1 X' m
( h" g0 X* V" P& Z+ b    //打印结果

$ g1 q* W! M# N& C4 q$ I1 ^" Z) Y    String jobName = "user defined streaming source";

' B( o) c+ G2 g% k    env.execute(jobName);

/ q6 ^/ V) J  E: N}
1 W% `- s8 `) H5 k' w</code></pre>
# F" t1 M( ]8 k3 f<p>在上面的例子中我们使用 filter 算子将原始流进行了拆分，输入数据第一个元素为 0 的数据和第一个元素为 1 的数据分别被写入到了 zeroStream 和 oneStream 中，然后把两个流进行了打印。</p>
0 l3 N$ w( @* v) i, j<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/Ciqc1F7CA2WAYbshAAKj494h86s723-20210223084827296.png" ></p>
<p>可以看到 zeroStream 和 oneStream 分别被打印出来。</p>
1 K$ w  d! N0 s! I& p1 [- P. S<p>Filter 的弊端是显而易见的，为了得到我们需要的流数据，需要多次遍历原始流，这样无形中浪费了我们集群的资源。</p>
4 G+ j+ J9 i& X" f8 i+ E" J<h4 id="split-分流">Split 分流</h4>
) M" Y/ K# ^8 b" F2 G6 E3 O9 x, I/ H9 E<p>Split 也是 Flink 提供给我们将流进行切分的方法，需要在 split 算子中定义 OutputSelector，然后重写其中的 select 方法，将不同类型的数据进行标记，最后对返回的 SplitStream 使用 select 方法将对应的数据选择出来。</p>
1 P0 B# e4 L' X$ |( G% k<p>我们来看下面的例子：</p>
: b  h) e6 @# X( b<p>复制代码</p>
<pre><code class="language-java">public static void main(String[] args) throws Exception {


3 q. i+ J  ^( f, b- e
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
$ u- Q: }8 z- b8 c
    //获取数据源

: o/ D! a: L. K5 \, Y. s' {    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
  L* \2 q6 K7 M0 ~- I
    data.add(new Tuple3&lt;&gt;(0,1,0));

- d9 O( T, \4 Q, x    data.add(new Tuple3&lt;&gt;(0,1,1));
- y+ E9 N6 {" B4 {. U
    data.add(new Tuple3&lt;&gt;(0,2,2));
# z, L- Y  X( t: t: h4 L- R* O
    data.add(new Tuple3&lt;&gt;(0,1,3));

    data.add(new Tuple3&lt;&gt;(1,2,5));
& T- Z3 i. \' C7 W1 \4 C
. r: i" Z7 m% s- E: l6 g    data.add(new Tuple3&lt;&gt;(1,2,9));

4 ]/ c! t- Y0 \! [' ?    data.add(new Tuple3&lt;&gt;(1,2,11));

    data.add(new Tuple3&lt;&gt;(1,2,13));
3 {6 @* H5 r3 r- X+ i
8 z/ }8 \1 F) a) G9 |( N
/ k( f" J/ y: V+ D4 M( e
( c* p& ^! B$ u1 s- U( v2 O0 {
9 [5 T' Z. K% h% u: r
$ ?% o7 ~* Q; g2 @/ B  I  o' V    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
' h8 S8 W6 m' K& u' J

/ ^/ r2 ?- x; |" b. S" S& s- P
8 Q3 K- S; z) y  m/ c& ^0 v5 m
6 G  r* H6 c" s9 m' a) K  w3 \
2 c3 p; S9 \  L! j7 {: }    SplitStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; splitStream = items.split(new OutputSelector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt;() {
* E% S  K3 j% X
8 i( |9 m8 N4 A% E: Z        @Override

        public Iterable&lt;String&gt; select(Tuple3&lt;Integer, Integer, Integer&gt; value) {

; [- W& c9 D$ F            List&lt;String&gt; tags = new ArrayList&lt;&gt;();

            if (value.f0 == 0) {
# d  L+ z1 M7 ]% P4 i7 c) `
% W/ p( J' E1 U/ S                tags.add("zeroStream");

5 g) R: [: W' N* S5 i7 z! G            } else if (value.f0 == 1) {

                tags.add("oneStream");

# t7 e4 z: y( L4 K            }
5 H( `+ W7 e# M) Y# Z* ?- {
6 e3 O7 u: j1 @, }  |            return tags;

        }
- p! s4 @0 B' ?" ?* z
    });

& j5 m# K+ y% x; {) l! l' Q

' f+ z1 b0 P; r7 e' e    splitStream.select("zeroStream").print();

. J2 [& e( o8 p3 r    splitStream.select("oneStream").printToErr();
! h  e' f# V& q- w9 P, X  D; e
0 b, N8 e! N  X  S# ~; k
, N: c/ k6 j! _$ B$ n
4 o% h2 ?* M' R# J7 h/ `! z    //打印结果
: K( z' J2 F2 L9 f2 I9 `
    String jobName = "user defined streaming source";
2 q& K) D0 Q6 M
& C3 g( k  r  y* ^7 L, r    env.execute(jobName);

0 z( L6 b- z; n9 @* q  c& K}
</code></pre>
<p>同样，我们把来源的数据使用 split 算子进行了切分，并且打印出结果。</p>
" ], q6 O8 T4 T( L! x<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA4aAbUSJAAG1LWNB3qw627-20210223084827377.png" ></p>
<p>但是要注意，使用 split 算子切分过的流，是不能进行二次切分的，假如把上述切分出来的 zeroStream 和 oneStream 流再次调用 split 切分，控制台会抛出以下异常。</p>
<p>复制代码</p>
& \$ D: v9 D4 k$ ?( H<pre><code class="language-java">Exception in thread "main" java.lang.IllegalStateException: Consecutive multiple splits are not supported. Splits are deprecated. Please use side-outputs.
</code></pre>
: H5 O. T# Y: u4 H4 L+ _5 l<p>这是什么原因呢？我们在源码中可以看到注释，该方式已经废弃并且建议使用最新的 SideOutPut 进行分流操作。</p>
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA6OAJ-JDAAIrh1JSAEo033-20210223084827602.png" ></p>
<h4 id="sideoutput-分流">SideOutPut 分流</h4>
<p>SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法，在使用 SideOutPut 时，需要按照以下步骤进行：</p>
* [( M# P, x9 I/ w1 U<ul>
8 b) H# C, d6 M' q6 ?1 D<li>定义 OutputTag</li>
( X# x! [7 v/ [7 D<li>调用特定函数进行数据拆分
, C/ w5 I" e- @( @  _<ul>
+ `1 m; V" s. D, V( K<li>ProcessFunction</li>
<li>KeyedProcessFunction</li>
6 z+ I, d& a9 _$ _; |6 a<li>CoProcessFunction</li>
' T" b# `+ m$ `# X6 _# S- @, J<li>KeyedCoProcessFunction</li>
<li>ProcessWindowFunction</li>
<li>ProcessAllWindowFunction</li>
</ul>
</li>
, K  P5 k0 L: T7 t8 s. j</ul>
<p>在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut：</p>
4 D: I* S/ V. g4 I. d" ~) }<p>复制代码</p>
& t4 B* r; }# n9 W5 g7 B1 k<pre><code class="language-java">public static void main(String[] args) throws Exception {
: v- C6 n' v: F$ H$ q


2 p5 T, J* \$ J( L6 O  c    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
: v. X: e% p6 _; l4 b
" R+ H0 [& q+ _; ]5 M    //获取数据源
: b3 Y* x9 d: |! D$ ]
    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();

9 J4 Y4 ^! h& W* @$ l    data.add(new Tuple3&lt;&gt;(0,1,0));

    data.add(new Tuple3&lt;&gt;(0,1,1));

3 f+ L, }# k, f- x- Z$ N6 _    data.add(new Tuple3&lt;&gt;(0,2,2));
+ `- e1 ]" U* A3 K. u* q: s
/ p3 h( k/ r  K/ k    data.add(new Tuple3&lt;&gt;(0,1,3));

8 ^+ F0 ]0 |6 ^& `% U    data.add(new Tuple3&lt;&gt;(1,2,5));
: ]3 e/ d) s9 S8 h
/ k( r$ Z9 g5 o9 A7 d/ e) k    data.add(new Tuple3&lt;&gt;(1,2,9));

4 f% V0 \7 Z& |# s    data.add(new Tuple3&lt;&gt;(1,2,11));
; @; l- T; Z6 S
    data.add(new Tuple3&lt;&gt;(1,2,13));





/ \+ F+ u1 ^/ k$ x) V6 s' Z    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
5 W2 m1 t7 _$ x. F1 P0 [

1 d% H0 I7 L, A# D; z
( l5 _+ M' m' Y. X/ p, s    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; zeroStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("zeroStream") {};

% q3 i( b  r' u- d  D, o5 l    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; oneStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("oneStream") {};

  D; Y/ s( W6 c' e

% u) j1 I) u& L. d2 M! G/ c
6 e0 L0 X3 r# ~
    SingleOutputStreamOperator&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; processStream= items.process(new ProcessFunction&lt;Tuple3&lt;Integer, Integer, Integer&gt;, Tuple3&lt;Integer, Integer, Integer&gt;&gt;() {
: p( _/ D; `+ ?) U1 E3 z' I
! M# ~  j3 z" |- {& C% |        @Override
% \- r. \0 ~6 s& Q* T9 e
. Z2 a4 A' ~+ |/ q- `+ e) h        public void processElement(Tuple3&lt;Integer, Integer, Integer&gt; value, Context ctx, Collector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; out) throws Exception {



            if (value.f0 == 0) {

                ctx.output(zeroStream, value);

            } else if (value.f0 == 1) {
) N! b9 a( ^0 i( b! y9 d8 Z
                ctx.output(oneStream, value);
' G% \1 z: V! d, S/ N- e4 f
            }

# D6 ~, _( P6 k! r3 M        }
1 W# ~/ Y2 z* W& E; Y; |! W" x
! c, ?( [; E- I* f    });

9 t" H. y* {& s9 R' U( W! m
- \. }; ~- k) U) k
8 \) N8 @7 H: r+ R! K  {    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; zeroSideOutput = processStream.getSideOutput(zeroStream);
$ j( [- d6 c" E+ Z7 D4 a: U- E, c
* a9 h1 \2 r* l0 V; N    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; oneSideOutput = processStream.getSideOutput(oneStream);

. E! j9 Z6 T7 j) I7 t/ i5 r

& x; F9 x! Y: Q: z/ H, Y; ^  s- o    zeroSideOutput.print();

    oneSideOutput.printToErr();
9 k# P4 j) s! i4 H5 Q, t- h: s
+ h& K  G  Y9 u9 c, I5 B
# F! H) J- |" Z3 b


    //打印结果
- w( ?- I+ @+ A& _9 ], L
& o, S" Q0 s+ n( T) [4 `+ O; L% {    String jobName = "user defined streaming source";

    env.execute(jobName);

0 S5 A9 K3 k* g}
</code></pre>
<p>可以看到，我们将流进行了拆分，并且成功打印出了结果。这里要注意，Flink 最新提供的 SideOutPut 方式拆分流是<strong>可以多次进行拆分</strong>的，无需担心会爆出异常。</p>
: ^6 b/ y( i: @5 K, ?% j" l<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CBMKAGHoUAAM-5UL5geg132-20210223084827698.png" ></p>
3 ^# P0 J& L+ z! ]& d+ C9 A<h3 id="总结">总结</h3>
- W. e" x& E: J; P) d; h<p>这一课时我们讲解了 Flink 的一个小的知识点，是我们生产实践中经常遇到的场景，Flink 在最新的版本中也推荐我们使用 SideOutPut 进行流的拆分。</p>
<blockquote>
<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
</blockquote>

; G3 K: ~( v& Z# n. S& ?! t