第10讲：Flink Side OutPut 分流

admin

<h4 id="flink系列文章">Flink系列文章</h4>
<ol>
' x7 G" x: ~. i<li><a  href="https://www.ikeguang.com/?p=1976">第01讲：Flink 的应用场景和架构模型</a></li>
, _7 o2 Z$ s. ~! P) `- u! s) k; h: r<li><a  href="https://www.ikeguang.com/?p=1977">第02讲：Flink 入门程序 WordCount 和 SQL 实现</a></li>
<li><a  href="https://www.ikeguang.com/?p=1978">第03讲：Flink 的编程模型与其他框架比较</a></li>
+ z; `1 Z) j! ~+ C. x3 ?<li><a  href="https://www.ikeguang.com/?p=1982">第04讲：Flink 常用的 DataSet 和 DataStream API</a></li>
<li><a  href="https://www.ikeguang.com/?p=1983">第05讲：Flink SQL &amp; Table 编程和案例</a></li>
/ m1 k5 }: D( w" o! q$ F<li><a  href="https://www.ikeguang.com/?p=1985">第06讲：Flink 集群安装部署和 HA 配置</a></li>
<li><a  href="https://www.ikeguang.com/?p=1986">第07讲：Flink 常见核心概念分析</a></li>
<li><a  href="https://www.ikeguang.com/?p=1987">第08讲：Flink 窗口、时间和水印</a></li>
( ]3 N6 S5 s, I  {5 D8 r% Q& H4 _<li><a  href="https://www.ikeguang.com/?p=1988">第09讲：Flink 状态与容错</a></li>
</ol>
2 T+ W/ b% }3 R1 K+ F<blockquote>
" I& X0 i8 a2 ?5 ~9 u<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
7 g8 _) x' L( q+ A& u</blockquote>
<p>这一课时将介绍 Flink 中提供的一个很重要的功能：旁路分流器。</p>
8 t4 `3 K$ o  i<h3 id="分流场景">分流场景</h3>
<p>我们在生产实践中经常会遇到这样的场景，需把输入源按照需要进行拆分，比如我期望把订单流按照金额大小进行拆分，或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢？</p>
<h3 id="分流的方法">分流的方法</h3>
" A% i0 \. U3 x! l<p>通常来说针对不同的场景，有以下三种办法进行流的拆分。</p>
; l, C8 U) {' |5 h$ G. m' W<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>
6 B, c+ X! ~$ ~* F2 g<p>来看下面的例子：</p>
<p>复制代码</p>
<pre><code class="language-java">public static void main(String[] args) throws Exception {
' o8 F& b* T7 N& m    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //获取数据源
    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
% W5 k' K; l) t% B    data.add(new Tuple3&lt;&gt;(0,1,0));
2 O9 ^- d' e# O/ r$ m    data.add(new Tuple3&lt;&gt;(0,1,1));
    data.add(new Tuple3&lt;&gt;(0,2,2));
    data.add(new Tuple3&lt;&gt;(0,1,3));
    data.add(new Tuple3&lt;&gt;(1,2,5));
    data.add(new Tuple3&lt;&gt;(1,2,9));
    data.add(new Tuple3&lt;&gt;(1,2,11));
    data.add(new Tuple3&lt;&gt;(1,2,13));

4 W; S- u0 k+ g0 A# n$ |% O8 X4 V    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
5 Y  t+ u8 f! K3 I$ C% c/ f+ q
+ N8 F" A2 Y6 K' C8 W
  s  P) e2 y, }
    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);

2 h9 O/ T' ^/ H
2 @: _+ V1 v+ {5 |% b1 X
    zeroStream.print();
, C8 g* W8 g( r+ g5 K5 z1 k! U* D
    oneStream.printToErr();
/ Y% L$ ~/ n2 ^) W1 }- |1 _1 \

# M' f* d( c4 b. ^" R5 j& X

, m) o5 y3 q' m3 S9 `
7 a: `3 m/ J/ `& z4 E$ S* s    //打印结果
4 r3 J9 N- D* t2 I# B
* i$ O  ^" W) a; [) p9 d( K    String jobName = "user defined streaming source";

3 @- Y- d2 l; i* v  {0 G    env.execute(jobName);

}
& F( q0 \9 O0 O7 g; q</code></pre>
<p>在上面的例子中我们使用 filter 算子将原始流进行了拆分，输入数据第一个元素为 0 的数据和第一个元素为 1 的数据分别被写入到了 zeroStream 和 oneStream 中，然后把两个流进行了打印。</p>
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/Ciqc1F7CA2WAYbshAAKj494h86s723-20210223084827296.png" ></p>
<p>可以看到 zeroStream 和 oneStream 分别被打印出来。</p>
<p>Filter 的弊端是显而易见的，为了得到我们需要的流数据，需要多次遍历原始流，这样无形中浪费了我们集群的资源。</p>
<h4 id="split-分流">Split 分流</h4>
<p>Split 也是 Flink 提供给我们将流进行切分的方法，需要在 split 算子中定义 OutputSelector，然后重写其中的 select 方法，将不同类型的数据进行标记，最后对返回的 SplitStream 使用 select 方法将对应的数据选择出来。</p>
/ Q% w5 v' N/ [$ w6 m7 ]! q<p>我们来看下面的例子：</p>
! {* u8 f7 v7 K<p>复制代码</p>
2 X. s: c( M/ P  a. C& h# m3 A<pre><code class="language-java">public static void main(String[] args) throws Exception {
" d% {: u& a  P- |0 @7 p
# ?% |6 |1 |8 q, G* P
3 u* g- _: |. E& \$ w0 }7 @
1 _6 R" t( j* F- r2 e3 j# R    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
5 o+ i( H  w% s0 g9 d
    //获取数据源

    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
" O6 D/ F' H1 B8 |9 s  i# P
    data.add(new Tuple3&lt;&gt;(0,1,0));

, a1 Q3 @7 |; r( M6 m4 t: v( Z    data.add(new Tuple3&lt;&gt;(0,1,1));

1 s) D- {  {$ J7 s( _) `1 }    data.add(new Tuple3&lt;&gt;(0,2,2));

& Q$ Q5 f; V% B    data.add(new Tuple3&lt;&gt;(0,1,3));

    data.add(new Tuple3&lt;&gt;(1,2,5));
: ]6 }5 j2 |/ S2 g
    data.add(new Tuple3&lt;&gt;(1,2,9));
' u! s& H: I- f( F
    data.add(new Tuple3&lt;&gt;(1,2,11));
% M- N, {' x! T; e" m3 ?+ z; Z
: O# d9 A/ {  N2 ]/ r+ N    data.add(new Tuple3&lt;&gt;(1,2,13));
! D# n5 b, N6 D! u. S0 @4 h- X1 p$ [

7 s1 p1 n$ h( H  r


  S8 s7 w( G* L  _' G    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);




$ I5 x( w$ `2 W2 n/ R
4 q; U5 m3 o1 x# X    SplitStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; splitStream = items.split(new OutputSelector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt;() {
; p2 F% k8 ~6 g7 w9 |
. D) t4 s+ n+ p7 w        @Override
& I9 e) c8 D$ d; u+ c
        public Iterable&lt;String&gt; select(Tuple3&lt;Integer, Integer, Integer&gt; value) {
6 l7 F9 r# j; I. l
            List&lt;String&gt; tags = new ArrayList&lt;&gt;();
5 U, j+ a- n; }7 w
$ V' Z$ {& s; [' p0 j2 w, F. {            if (value.f0 == 0) {
8 ^/ X8 Z  p5 ], R6 E: ?6 t
' e6 H( H3 Z  P/ d/ j7 P                tags.add("zeroStream");
' O1 q1 Z: n+ B) v& Y
: `6 X1 I) m) p- o            } else if (value.f0 == 1) {
8 y& ?7 v1 f2 P5 N
                tags.add("oneStream");

            }

            return tags;

. m' B" K+ h) Z% N/ V  l* Z        }

8 W  F2 Y& \7 s    });

6 E7 ~3 F) p) U9 ~' x
8 Y. f% ?: N5 ]7 j9 R
    splitStream.select("zeroStream").print();
/ ~7 p( f/ q% M2 L* n9 [% W: }
" O$ q# ~6 J3 Q; `( m3 b    splitStream.select("oneStream").printToErr();

* S! J) }; F: i& G) T. m
) @8 y0 H, A% k$ Z0 L; ~9 i5 o( G; ?
    //打印结果
4 v% H! {0 i7 ~/ w
! |; ]1 j' e& ?: J    String jobName = "user defined streaming source";
, V" ^' @4 E/ Y. Y: E
    env.execute(jobName);
( x6 Q- H) Q5 l% x7 U, S' p4 W
: B' Z: n1 `, K}
</code></pre>
<p>同样，我们把来源的数据使用 split 算子进行了切分，并且打印出结果。</p>
- C% j; o$ M0 i4 W# u<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA4aAbUSJAAG1LWNB3qw627-20210223084827377.png" ></p>
7 M3 Z7 A( x3 K* x<p>但是要注意，使用 split 算子切分过的流，是不能进行二次切分的，假如把上述切分出来的 zeroStream 和 oneStream 流再次调用 split 切分，控制台会抛出以下异常。</p>
9 @" e. J# U( o! w5 i: o; {3 U3 q7 d<p>复制代码</p>
2 K( Z4 \' V: L' {4 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>
# K3 q* g2 c3 x7 ]<p>这是什么原因呢？我们在源码中可以看到注释，该方式已经废弃并且建议使用最新的 SideOutPut 进行分流操作。</p>
' U$ P  y+ X# s3 s/ R: e& b<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA6OAJ-JDAAIrh1JSAEo033-20210223084827602.png" ></p>
3 Y8 R- P# I  {" a2 {<h4 id="sideoutput-分流">SideOutPut 分流</h4>
3 z# n- E1 j3 x7 x: e<p>SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法，在使用 SideOutPut 时，需要按照以下步骤进行：</p>
<ul>
<li>定义 OutputTag</li>
<li>调用特定函数进行数据拆分
<ul>
<li>ProcessFunction</li>
# r+ g% i. N& m) O9 W+ U<li>KeyedProcessFunction</li>
<li>CoProcessFunction</li>
/ U6 \8 N7 T$ ]1 Y/ B<li>KeyedCoProcessFunction</li>
* H# R& ?! O$ }% p2 p* T# V<li>ProcessWindowFunction</li>
<li>ProcessAllWindowFunction</li>
0 Y0 n7 L6 F8 R* F</ul>
) j  G8 Q, O! i# r! h! r</li>
</ul>
- t( z( N" i2 Y) O4 P* Q$ i<p>在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut：</p>
% `% g, w4 D3 `! Y- h9 ]* d, a<p>复制代码</p>
' M! F$ W. N: Z2 t2 ~5 }. }<pre><code class="language-java">public static void main(String[] args) throws Exception {
1 w) U6 U  h  |$ k; d

9 A: R* s* t9 G$ p+ q
  K. Z1 A# S2 M; T    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    //获取数据源

    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
3 f1 G% t$ A- p& T+ D
( Q9 `9 c; _' ^/ l( d    data.add(new Tuple3&lt;&gt;(0,1,0));
/ ~. M  {1 p9 j: i. [7 `, n8 n, x
    data.add(new Tuple3&lt;&gt;(0,1,1));

. w- T% a7 ]7 I) S    data.add(new Tuple3&lt;&gt;(0,2,2));

    data.add(new Tuple3&lt;&gt;(0,1,3));
! E) ?8 ^  Q- Y5 Y3 @3 P) [2 d' A
3 i4 d. a7 ~2 b7 n- H    data.add(new Tuple3&lt;&gt;(1,2,5));

    data.add(new Tuple3&lt;&gt;(1,2,9));
. t9 D2 O( K& A5 Q7 j& W1 R/ c
    data.add(new Tuple3&lt;&gt;(1,2,11));

: T; H3 A* t6 y) P    data.add(new Tuple3&lt;&gt;(1,2,13));

" T3 o+ e' A# r+ g$ o
! T6 k4 N5 v* F; z& Z# C! f+ c
" k3 g: Z) ^5 L0 h8 U. o7 a5 q
: {; Y8 c+ m3 _: L
: Q& J* {: G' a* V( O1 L    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
: q, y9 f1 `6 u; i5 l


    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; zeroStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("zeroStream") {};
1 C2 y- u! D3 p% s% F5 }: |
    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; oneStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("oneStream") {};

8 J3 }; }! O" o! \* p


+ N2 m( K+ J0 c
    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;() {
( H$ ^8 F1 T) L
        @Override
" g$ g( _. f! |9 I, w& f* C
  |0 a* @! }- _0 H$ z        public void processElement(Tuple3&lt;Integer, Integer, Integer&gt; value, Context ctx, Collector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; out) throws Exception {
8 J2 \$ c0 F% D5 B& v7 P1 x% W$ g


3 R; M) \8 t& X4 H- q% u; x( e* b            if (value.f0 == 0) {

                ctx.output(zeroStream, value);

            } else if (value.f0 == 1) {

                ctx.output(oneStream, value);

6 E, w( V9 S5 m) E2 p' f9 [            }
; s# g( K2 e, G
        }
4 T6 @6 g6 s0 f
2 W+ W4 |1 B3 }4 g0 _; ]" C$ f    });


# e) z" s: V% h7 z3 D- N" S
& _4 m, b4 b( ?# \- ~9 W  K# n    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; zeroSideOutput = processStream.getSideOutput(zeroStream);
9 F' `& K0 t" b6 |7 o2 p' v3 n
    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; oneSideOutput = processStream.getSideOutput(oneStream);


* d' P9 n) n4 ]7 I. U
( `3 T# d  W$ V    zeroSideOutput.print();
0 u3 ~: T% H: w
    oneSideOutput.printToErr();
/ R1 D1 e! X7 g" H# \' @
% J# h2 o1 X/ Z- Z# D% I
2 u2 V; x! ^! Z+ Y- y: s' o& ?9 b. ^. v
. T! i* q  {% |8 M5 M
9 u" m9 Z2 |- W. ^4 c2 ^5 \
    //打印结果

    String jobName = "user defined streaming source";
% W# o9 H+ w! D7 [& D! `
    env.execute(jobName);
. @. ^4 C# ]* H4 f# M( e/ V
}
</code></pre>
2 C. R" f" B: e- A' x$ z<p>可以看到，我们将流进行了拆分，并且成功打印出了结果。这里要注意，Flink 最新提供的 SideOutPut 方式拆分流是<strong>可以多次进行拆分</strong>的，无需担心会爆出异常。</p>
, L2 I; t. x1 _. t<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CBMKAGHoUAAM-5UL5geg132-20210223084827698.png" ></p>
<h3 id="总结">总结</h3>
' B* ?5 I: m# c/ A+ m+ a<p>这一课时我们讲解了 Flink 的一个小的知识点，是我们生产实践中经常遇到的场景，Flink 在最新的版本中也推荐我们使用 SideOutPut 进行流的拆分。</p>
<blockquote>
<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
2 [5 C! v  ]" M0 I</blockquote>
. x4 z3 c2 U( H- P* u7 O