第10讲：Flink Side OutPut 分流

admin

<h4 id="flink系列文章">Flink系列文章</h4>
) F% y# M$ q' x' e& [<ol>
2 e6 u, B6 u: h8 X% `<li><a  href="https://www.ikeguang.com/?p=1976">第01讲：Flink 的应用场景和架构模型</a></li>
$ P4 B) \' S9 d& \, F8 ]8 n& }<li><a  href="https://www.ikeguang.com/?p=1977">第02讲：Flink 入门程序 WordCount 和 SQL 实现</a></li>
. l4 U% }% U3 K<li><a  href="https://www.ikeguang.com/?p=1978">第03讲：Flink 的编程模型与其他框架比较</a></li>
' n: [+ ?& ]$ p( P5 H+ E; j<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>
9 [9 e% v3 t3 d, L$ L& k, b<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>
5 Q1 X. I1 Y* a. w' t3 I" H3 z<li><a  href="https://www.ikeguang.com/?p=1987">第08讲：Flink 窗口、时间和水印</a></li>
' r- S4 U+ \2 U0 O; p<li><a  href="https://www.ikeguang.com/?p=1988">第09讲：Flink 状态与容错</a></li>
</ol>
+ b# x9 J' P0 a& p# |<blockquote>
<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
8 G( t  \& A7 m4 }" Y</blockquote>
<p>这一课时将介绍 Flink 中提供的一个很重要的功能：旁路分流器。</p>
<h3 id="分流场景">分流场景</h3>
<p>我们在生产实践中经常会遇到这样的场景，需把输入源按照需要进行拆分，比如我期望把订单流按照金额大小进行拆分，或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢？</p>
<h3 id="分流的方法">分流的方法</h3>
<p>通常来说针对不同的场景，有以下三种办法进行流的拆分。</p>
' K! W+ N3 e* m<h4 id="filter-分流">Filter 分流</h4>
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CAy6ADUaXAACSFUbdpuA911-20210223084827182.png" ></p>
: v& U) q; W6 e; T0 F2 D<p>Filter 方法我们在第 04 课时中（Flink 常用的 DataSet 和 DataStream API）讲过，这个算子用来根据用户输入的条件进行过滤，每个元素都会被 filter() 函数处理，如果 filter() 函数返回 true 则保留，否则丢弃。那么用在分流的场景，我们可以做多次 filter，把我们需要的不同数据生成不同的流。</p>
3 @! g2 W% W7 r6 S& t# |<p>来看下面的例子：</p>
<p>复制代码</p>
<pre><code class="language-java">public static void main(String[] args) throws Exception {
" g- x; I8 g( d3 Z& u    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //获取数据源
    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));
) S2 |1 Y( B8 n    data.add(new Tuple3&lt;&gt;(0,1,3));
2 w3 Z' Z; q! V4 G' o) A; c% h% T    data.add(new Tuple3&lt;&gt;(1,2,5));
    data.add(new Tuple3&lt;&gt;(1,2,9));
7 x/ \; h$ M7 f! S- C    data.add(new Tuple3&lt;&gt;(1,2,11));
    data.add(new Tuple3&lt;&gt;(1,2,13));

    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
/ F  w. Q  P  R) I/ @


    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);
/ q0 t# x* V/ j) P, W
    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);
( d! R, L- L& y' Z5 x


    zeroStream.print();

    oneStream.printToErr();

* L: J' {' T- E

+ U8 F: }; k! @" F9 }+ C6 a5 A' W
2 p# A7 G9 s( a3 y9 L4 l1 s
    //打印结果
8 T) [1 U. U3 a; y0 Z
4 b  m5 j) H# l! W" n) }$ Z2 w    String jobName = "user defined streaming source";
1 Y4 s! t3 K  C, g
    env.execute(jobName);

/ G  i" q# @9 X}
. f' `# Z7 `* z2 |; T9 q# K</code></pre>
<p>在上面的例子中我们使用 filter 算子将原始流进行了拆分，输入数据第一个元素为 0 的数据和第一个元素为 1 的数据分别被写入到了 zeroStream 和 oneStream 中，然后把两个流进行了打印。</p>
  I# b. ]' b* s! J! F<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/Ciqc1F7CA2WAYbshAAKj494h86s723-20210223084827296.png" ></p>
: M& e. T9 }7 ]" `6 Y- l* j<p>可以看到 zeroStream 和 oneStream 分别被打印出来。</p>
' P0 K( g5 K* n- v4 s: O- a, {, B5 `5 u<p>Filter 的弊端是显而易见的，为了得到我们需要的流数据，需要多次遍历原始流，这样无形中浪费了我们集群的资源。</p>
<h4 id="split-分流">Split 分流</h4>
<p>Split 也是 Flink 提供给我们将流进行切分的方法，需要在 split 算子中定义 OutputSelector，然后重写其中的 select 方法，将不同类型的数据进行标记，最后对返回的 SplitStream 使用 select 方法将对应的数据选择出来。</p>
# w' @$ F' N- \3 E% G5 J<p>我们来看下面的例子：</p>
' g5 z$ C) M4 O6 F. g- A<p>复制代码</p>
, w7 b& H- s& _, G& g5 r" e' s* ^<pre><code class="language-java">public static void main(String[] args) throws Exception {
' }4 J# H2 K+ K' M+ R


+ E' h% \5 M& O0 V/ L2 X& ]( A    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  {  _: D% i5 S: E: j8 a2 _    //获取数据源
' F( }  @& f  \, D
    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
, {) t9 C- z9 [6 N  K* `$ h( W9 t
5 ]( h  C$ S. A    data.add(new Tuple3&lt;&gt;(0,1,0));

8 `; W2 a1 G, E3 f$ P( k& b    data.add(new Tuple3&lt;&gt;(0,1,1));
$ l: [: U/ X# `' b1 N3 Z' V) O
& F' @# B0 p, Q  w* |    data.add(new Tuple3&lt;&gt;(0,2,2));
$ O5 [, J8 M* v7 G9 S1 @9 j
% p3 \1 @7 d* R$ v' Q2 r    data.add(new Tuple3&lt;&gt;(0,1,3));
8 d2 j3 G7 V# w& }) [
- A- [- ?5 u. ^) D    data.add(new Tuple3&lt;&gt;(1,2,5));
2 l  Z, P' W" |+ [- z
9 R* m4 e1 ]3 j8 {7 U. R. t    data.add(new Tuple3&lt;&gt;(1,2,9));

    data.add(new Tuple3&lt;&gt;(1,2,11));
+ X- n% A& x; V- l$ P3 ~- ~
    data.add(new Tuple3&lt;&gt;(1,2,13));



8 o& F4 K% f% n! B5 ?
4 W; n9 `: r+ S% f& E' W* m( j
    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);

  ~1 t& l9 L4 U
* f! s+ q7 n3 D4 s6 W& Q9 }
0 @3 B, b5 A+ V, c8 \% k
0 f$ C! L+ v2 F3 G
3 p0 f+ ~5 d" O. C    SplitStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; splitStream = items.split(new OutputSelector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt;() {

        @Override

        public Iterable&lt;String&gt; select(Tuple3&lt;Integer, Integer, Integer&gt; value) {
; H9 M; F$ Q% B0 H
* q- j' o: d: J7 Q. E            List&lt;String&gt; tags = new ArrayList&lt;&gt;();
8 u! ^0 o' J  b2 J/ `* L
* Q' I# `# K- O2 h4 O& W            if (value.f0 == 0) {
+ }* `2 d) f" [6 Y1 [
                tags.add("zeroStream");
# _1 e. k* S2 q) v
6 f5 E8 t% H3 E7 m# Y            } else if (value.f0 == 1) {
! k4 F2 g) p- u% X1 R0 a0 c! K1 G
" c. Y1 j. @9 u) Z                tags.add("oneStream");

            }
. ], `  x4 c. g9 u) h& x
            return tags;
# B) S3 U" X3 s' @- T. D
& m! a. d" E4 D8 e) c4 c" z        }
2 M. e! q; S, \
* ]% s; T) @' l3 v    });
! t8 r  d8 p3 \1 i  G4 e2 l1 g
- p3 j0 z$ ^  |* T
0 P& K  T* e) S- ^1 [
    splitStream.select("zeroStream").print();

/ o6 M% X- r1 w$ X; G+ x$ `1 i    splitStream.select("oneStream").printToErr();
, a$ T6 d$ }2 ?" h5 ~. y

' `; r2 c/ N/ W) T. t1 r7 q7 u
0 W6 t* ~+ c  c6 [7 @# s! |    //打印结果
+ l3 ?# _7 m! r9 k$ F
8 ]* D5 c! z) d% g+ k    String jobName = "user defined streaming source";
, \) r" O% K; d8 @; A
    env.execute(jobName);

2 _2 s3 C  w% V& C}
. U% @. K8 O4 m</code></pre>
<p>同样，我们把来源的数据使用 split 算子进行了切分，并且打印出结果。</p>
1 j) p, S7 b9 ~0 [4 f) ]9 a<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA4aAbUSJAAG1LWNB3qw627-20210223084827377.png" ></p>
<p>但是要注意，使用 split 算子切分过的流，是不能进行二次切分的，假如把上述切分出来的 zeroStream 和 oneStream 流再次调用 split 切分，控制台会抛出以下异常。</p>
* }! i, B# E& R7 p<p>复制代码</p>
; V8 [  ?- [* L. d<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>
<p>这是什么原因呢？我们在源码中可以看到注释，该方式已经废弃并且建议使用最新的 SideOutPut 进行分流操作。</p>
1 i7 P& O( Y/ U<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA6OAJ-JDAAIrh1JSAEo033-20210223084827602.png" ></p>
* m6 S5 W  Q3 q( i1 X) l- n<h4 id="sideoutput-分流">SideOutPut 分流</h4>
* P4 L8 K" S, t8 l6 p<p>SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法，在使用 SideOutPut 时，需要按照以下步骤进行：</p>
1 n9 C" v/ E6 g& M: g! j) f<ul>
8 e, v/ d! r9 u<li>定义 OutputTag</li>
. E, v* a. [; z1 s2 Q# S<li>调用特定函数进行数据拆分
; o0 w+ [) Z. }# A# d7 ?<ul>
<li>ProcessFunction</li>
0 y6 O! W5 H& B1 U8 m" W; u<li>KeyedProcessFunction</li>
<li>CoProcessFunction</li>
: s- W% i+ c' U% c, Q& p! X<li>KeyedCoProcessFunction</li>
/ G( O0 p, }' c, O! d3 B- c9 d<li>ProcessWindowFunction</li>
+ W0 {- r$ n/ E& O  L<li>ProcessAllWindowFunction</li>
) V' a6 E! y- \2 Y, Y8 @</ul>
</li>
( ~" U4 ?" C0 N( z6 {</ul>
<p>在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut：</p>
<p>复制代码</p>
% u) c) j7 t0 Z<pre><code class="language-java">public static void main(String[] args) throws Exception {
: V5 {8 `  [, r: `( _. n


' p2 v4 \9 i  O" I  R7 A    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
6 Y9 F' p9 H- s
    //获取数据源

3 t5 H; E6 M4 V3 ?* V    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();

* L- t1 J; S/ |0 O$ z$ i  r9 o    data.add(new Tuple3&lt;&gt;(0,1,0));
6 h0 Y" D. J+ L0 ~( ^4 I
    data.add(new Tuple3&lt;&gt;(0,1,1));
" \/ N$ Z% p; R7 e% i$ {
    data.add(new Tuple3&lt;&gt;(0,2,2));

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

- W. t/ L, K/ j( g& _5 |" Z" c    data.add(new Tuple3&lt;&gt;(1,2,5));
% U7 r- c- M7 d" G8 {1 J
2 z* X3 q! {0 O  t    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));

! M8 _5 u$ G; X5 o
, {/ Q4 l# O# c7 @, A8 Y) c; s0 [
1 d% N( V$ r$ ]# M* t- w
9 c! G% R' }% n) f8 m& T; y1 ]0 w
    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);

" _4 N9 C. S. L1 X0 q
2 G  U; w, g( Y9 c: B! e$ L
    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; zeroStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("zeroStream") {};
9 ]6 S" G5 w, L% m1 x% b# j/ m$ n
    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; oneStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("oneStream") {};


* u5 W* K8 V: \% v' x' ~5 s# M


  ^0 }7 L( d9 k    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;() {
$ t) d1 @! a" I- @4 S+ N- S
7 ?/ L5 E- D& U4 `$ ?& h5 r/ X3 z        @Override
) A% Y: z  F# |6 L& D7 i8 E: P- j
- ^2 d8 k6 a, M  U        public void processElement(Tuple3&lt;Integer, Integer, Integer&gt; value, Context ctx, Collector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; out) throws Exception {

9 Q$ M  W0 o" \9 W' ?

            if (value.f0 == 0) {
/ P7 O  @1 d+ I0 P3 |! ]
* {* j0 ~) Y: b5 ?8 V) @% B/ Q" a                ctx.output(zeroStream, value);
+ L  h3 b4 U( I8 g$ N1 r' ~
; q; d( ]: ?  ?: B+ i            } else if (value.f0 == 1) {

                ctx.output(oneStream, value);

            }

3 e  ]$ z# j, ^* j8 S  s+ C; ^1 O        }
. T4 }. q1 S" I7 `& ^% G
    });
+ H2 g, x3 D  M0 T) g/ k
3 d$ i  G. a7 V0 r0 |* k3 Q

    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; zeroSideOutput = processStream.getSideOutput(zeroStream);
& \$ E: d( x4 M" n6 |, X0 o
    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; oneSideOutput = processStream.getSideOutput(oneStream);


4 B' Y7 j) X$ C" @/ t4 `6 Q
    zeroSideOutput.print();

    oneSideOutput.printToErr();

" L  C0 u  l2 B* l* I1 w/ s


* H1 {! {: a5 h- _- U  N
4 I' a2 G, ^  x    //打印结果

' a) i" x* N/ a/ c    String jobName = "user defined streaming source";
8 p2 q% F/ u/ ?6 H' Q) h
    env.execute(jobName);
6 F5 i5 ~) G8 A. V7 W% E
0 [9 w# v1 b7 {/ O7 T7 l}
</code></pre>
<p>可以看到，我们将流进行了拆分，并且成功打印出了结果。这里要注意，Flink 最新提供的 SideOutPut 方式拆分流是<strong>可以多次进行拆分</strong>的，无需担心会爆出异常。</p>
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CBMKAGHoUAAM-5UL5geg132-20210223084827698.png" ></p>
<h3 id="总结">总结</h3>
<p>这一课时我们讲解了 Flink 的一个小的知识点，是我们生产实践中经常遇到的场景，Flink 在最新的版本中也推荐我们使用 SideOutPut 进行流的拆分。</p>
" Y" T  l' ]) K) I# e2 G9 {- t<blockquote>
<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
( z' v7 V% f& C</blockquote>
, B' V9 G9 r, V4 T; a* I$ o* g" G+ e
- l, v% t( f* ?) S