第10讲：Flink Side OutPut 分流

admin

<h4 id="flink系列文章">Flink系列文章</h4>
<ol>
) a' _* @7 [; y  M<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>
<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>
3 y/ K6 Z! B  \/ T/ |; n  X<li><a  href="https://www.ikeguang.com/?p=1983">第05讲：Flink SQL &amp; Table 编程和案例</a></li>
<li><a  href="https://www.ikeguang.com/?p=1985">第06讲：Flink 集群安装部署和 HA 配置</a></li>
5 ?5 `7 u" ~5 I; h% q) R( m<li><a  href="https://www.ikeguang.com/?p=1986">第07讲：Flink 常见核心概念分析</a></li>
/ Z" J# ?5 s; F& I<li><a  href="https://www.ikeguang.com/?p=1987">第08讲：Flink 窗口、时间和水印</a></li>
- h; [# Z- l5 L+ N) Q' }! d<li><a  href="https://www.ikeguang.com/?p=1988">第09讲：Flink 状态与容错</a></li>
</ol>
$ }* o$ p* V& U( H) T<blockquote>
  S3 V4 g/ l% ?# h8 B<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
" ?% S& @) U2 R</blockquote>
! I  o# \3 S( A) K- X! S( H<p>这一课时将介绍 Flink 中提供的一个很重要的功能：旁路分流器。</p>
<h3 id="分流场景">分流场景</h3>
( M" l; l; u! z. ~9 X' _8 G% w<p>我们在生产实践中经常会遇到这样的场景，需把输入源按照需要进行拆分，比如我期望把订单流按照金额大小进行拆分，或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢？</p>
<h3 id="分流的方法">分流的方法</h3>
. C; z/ }4 _% `, g<p>通常来说针对不同的场景，有以下三种办法进行流的拆分。</p>
% t$ @1 ]; i- t3 W3 [1 C  K+ O1 |<h4 id="filter-分流">Filter 分流</h4>
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CAy6ADUaXAACSFUbdpuA911-20210223084827182.png" ></p>
& W8 a! z/ p- V$ S, M8 V; h6 g<p>Filter 方法我们在第 04 课时中（Flink 常用的 DataSet 和 DataStream API）讲过，这个算子用来根据用户输入的条件进行过滤，每个元素都会被 filter() 函数处理，如果 filter() 函数返回 true 则保留，否则丢弃。那么用在分流的场景，我们可以做多次 filter，把我们需要的不同数据生成不同的流。</p>
0 }2 k9 i. ~$ h& Y' X/ _# [<p>来看下面的例子：</p>
" F$ `" z& L1 Y0 M5 L<p>复制代码</p>
" [. x4 Q/ T# G( M! g/ N<pre><code class="language-java">public static void main(String[] args) throws Exception {
    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));
/ Q1 O( {6 g& p4 {    data.add(new Tuple3&lt;&gt;(0,1,1));
    data.add(new Tuple3&lt;&gt;(0,2,2));
+ ]* H/ J0 D% ?" Z, f+ \    data.add(new Tuple3&lt;&gt;(0,1,3));
  Q8 Y5 W, o9 ^% ^, {    data.add(new Tuple3&lt;&gt;(1,2,5));
    data.add(new Tuple3&lt;&gt;(1,2,9));
+ F* l- Y" z7 [! M% J  ], n    data.add(new Tuple3&lt;&gt;(1,2,11));
    data.add(new Tuple3&lt;&gt;(1,2,13));
  O; v. c# A  M
9 n5 a0 D/ W0 q" `6 G    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
( q$ `/ `$ V  k' g1 ~


1 ^8 j2 X! q, m% F4 n0 Q    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);
# t8 y0 R* v4 V: A
    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 N. ]% P) V1 K. `4 |( M+ B
) K: y+ s, P( p8 K) o' V) h
/ Z+ N- J- H$ @' K% Q$ `3 ]
    zeroStream.print();
/ N. b2 R, C0 v3 X5 J1 M1 T! j5 e
    oneStream.printToErr();

$ x" e( b0 ~0 {. \' \9 y. t
1 |% ~; B; d4 H9 m( z- ]* }


    //打印结果

- n5 \3 _* h: B) P& @! E    String jobName = "user defined streaming source";
  u+ F" j# \" u8 X
( T3 v$ Y, x- f: o! ]! _* h/ o    env.execute(jobName);
5 S0 a& d% _5 Y6 e4 N
}
* V9 v/ [' u( u! F</code></pre>
0 x6 l3 C& `& H) l<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>
  w) K8 {% a% m2 u<p>Filter 的弊端是显而易见的，为了得到我们需要的流数据，需要多次遍历原始流，这样无形中浪费了我们集群的资源。</p>
<h4 id="split-分流">Split 分流</h4>
<p>Split 也是 Flink 提供给我们将流进行切分的方法，需要在 split 算子中定义 OutputSelector，然后重写其中的 select 方法，将不同类型的数据进行标记，最后对返回的 SplitStream 使用 select 方法将对应的数据选择出来。</p>
% Z. _4 q/ [5 B. W# `<p>我们来看下面的例子：</p>
<p>复制代码</p>
3 F% Z4 b5 }* [3 F3 m<pre><code class="language-java">public static void main(String[] args) throws Exception {
: G0 R: o+ X3 o
3 f4 O; E5 ^. ~: Z! t3 z

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    //获取数据源
, M$ {, e0 F- V+ D% @3 X; D
    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
8 M+ c  I  {: Z2 ?  J; Y( `
    data.add(new Tuple3&lt;&gt;(0,1,0));
$ w7 R$ y% u5 J9 [3 G2 ~
; }( v( b) `$ v! u# a. ?    data.add(new Tuple3&lt;&gt;(0,1,1));
# P; ?! B# U/ m2 D
6 a6 P2 z; ~7 [+ p  B; Q    data.add(new Tuple3&lt;&gt;(0,2,2));

% n& ?2 X# Z. t" m$ }: f* ]    data.add(new Tuple3&lt;&gt;(0,1,3));

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

( z+ h/ ]8 O( s' P; ^    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));

6 @2 C+ x3 \2 k2 y) }* M" c
$ Q, Q/ m! V/ s$ K8 U


; q+ @% _6 \  H0 H  v' _    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);
' _% _1 ^9 X& N4 g  ~5 k$ S


8 p+ [" |" l2 S+ b4 S9 v

$ h7 n& ^# G$ J6 V% ^    SplitStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; splitStream = items.split(new OutputSelector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt;() {

5 Z0 v# f3 J. X+ i# K" I4 [' H        @Override

        public Iterable&lt;String&gt; select(Tuple3&lt;Integer, Integer, Integer&gt; value) {
4 P6 A/ P3 u1 `6 d2 a' q! \2 x; G, ]
0 ~+ T* V1 M- H2 J1 {            List&lt;String&gt; tags = new ArrayList&lt;&gt;();
+ v3 M1 y; r  {2 i
            if (value.f0 == 0) {

                tags.add("zeroStream");
9 e8 U$ F. S* z9 s% t: ?/ ]
, |2 ~% Y/ W$ f- T+ @) I            } else if (value.f0 == 1) {

                tags.add("oneStream");

( a) |, r$ ~: ^( }            }
" |% N/ x+ }, \  X: I, H
. t& S# R. k) ?! @! }            return tags;

( j$ {, S$ z! o        }
8 b9 O7 r) P, p! F1 ~1 s% q5 D% w
    });

9 D: M3 _* P) C7 K/ k% F" c6 Y
2 z. [5 @5 A$ `
' b* ]' m: J( z2 z: O    splitStream.select("zeroStream").print();
- e+ k; ~- G" }
    splitStream.select("oneStream").printToErr();


7 f# \2 g. d4 v+ X, v
6 Z. ?4 ?0 d8 u( f: S, ?    //打印结果
& e( U8 }) c( }6 J: ^: A
9 a6 P- G, U; w  e    String jobName = "user defined streaming source";

    env.execute(jobName);

}
/ l; F$ [* j, P- x; V</code></pre>
<p>同样，我们把来源的数据使用 split 算子进行了切分，并且打印出结果。</p>
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA4aAbUSJAAG1LWNB3qw627-20210223084827377.png" ></p>
<p>但是要注意，使用 split 算子切分过的流，是不能进行二次切分的，假如把上述切分出来的 zeroStream 和 oneStream 流再次调用 split 切分，控制台会抛出以下异常。</p>
<p>复制代码</p>
0 J: Z7 ^% x* y& F" c<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.
. w! P$ `! c+ g1 r+ @% F6 S</code></pre>
+ H1 V, T  @2 ]0 M<p>这是什么原因呢？我们在源码中可以看到注释，该方式已经废弃并且建议使用最新的 SideOutPut 进行分流操作。</p>
: ]3 ~& H5 s% q<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA6OAJ-JDAAIrh1JSAEo033-20210223084827602.png" ></p>
6 G! V) g/ q" v' X9 W<h4 id="sideoutput-分流">SideOutPut 分流</h4>
<p>SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法，在使用 SideOutPut 时，需要按照以下步骤进行：</p>
$ }4 f4 Q/ \& n3 r<ul>
# k: I# y; Q( I9 n# u% |<li>定义 OutputTag</li>
<li>调用特定函数进行数据拆分
" R7 x2 h3 N; T<ul>
" b2 s3 h( _) j" b<li>ProcessFunction</li>
<li>KeyedProcessFunction</li>
<li>CoProcessFunction</li>
" O% W& m: q9 w# G<li>KeyedCoProcessFunction</li>
<li>ProcessWindowFunction</li>
<li>ProcessAllWindowFunction</li>
. Z# X, Q9 |7 Q' T) B! {</ul>
/ C+ }) Z  s, m4 d: Y</li>
</ul>
<p>在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut：</p>
<p>复制代码</p>
<pre><code class="language-java">public static void main(String[] args) throws Exception {

/ P7 l9 z0 y! q$ [) i* g8 Y
4 T3 z: t3 W. h
2 q3 [/ O( `2 W4 B8 q* R/ ]) |+ }4 m    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
* W$ T" S0 h6 J7 y3 N
    //获取数据源

    List data = new ArrayList&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;();
) F$ e/ f2 k' D  z
$ e2 V. Z) E; Q& [    data.add(new Tuple3&lt;&gt;(0,1,0));

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

2 D; I# _: F, l$ Q* i5 {9 o% m    data.add(new Tuple3&lt;&gt;(0,2,2));

. h# |$ {, a& b6 d% z6 e    data.add(new Tuple3&lt;&gt;(0,1,3));

    data.add(new Tuple3&lt;&gt;(1,2,5));
! @4 a9 L3 N  |8 J+ `# V; x3 c5 f
1 I: G0 A" l9 }( x/ C  R    data.add(new Tuple3&lt;&gt;(1,2,9));
6 s1 O2 B; j% q; J
    data.add(new Tuple3&lt;&gt;(1,2,11));
8 L$ y' K1 q0 [4 I  ]
4 u  d8 h% Z) [  \5 g    data.add(new Tuple3&lt;&gt;(1,2,13));
- F% @* u# [# H5 v# t




    DataStreamSource&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; items = env.fromCollection(data);



    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; zeroStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("zeroStream") {};

6 V/ e$ M7 x3 ^    OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt; oneStream = new OutputTag&lt;Tuple3&lt;Integer,Integer,Integer&gt;&gt;("oneStream") {};


! o# f% ?  e# @* m/ f
. E4 I9 F% g9 z, ^9 D
# r0 l* s2 A) j
- A$ a1 k% m: [7 O9 j1 o7 `1 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;() {

        @Override

        public void processElement(Tuple3&lt;Integer, Integer, Integer&gt; value, Context ctx, Collector&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; out) throws Exception {

+ ]& `- X. P* ^5 i: k. I5 D

            if (value.f0 == 0) {

) i# d9 t$ n2 z8 d- F                ctx.output(zeroStream, value);

, b) r7 ^! Q$ T2 n  w& X" x            } else if (value.f0 == 1) {
" c, V7 }2 R9 Y, G' U
) b7 Z$ d: W! J- d: C  P  t                ctx.output(oneStream, value);
2 F( N9 b4 f4 z1 E
% R$ P5 K* ]9 V. B( h* w            }

        }
$ H  _3 k8 t" ^/ J# v" @+ m
) r, R1 _) E- R. z. |    });



* H; ~7 t* J5 [' l1 I, J* o    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; zeroSideOutput = processStream.getSideOutput(zeroStream);

    DataStream&lt;Tuple3&lt;Integer, Integer, Integer&gt;&gt; oneSideOutput = processStream.getSideOutput(oneStream);

5 q% L! D) o: K+ u

    zeroSideOutput.print();
3 L& R. N' r3 M2 u) l2 s/ k
    oneSideOutput.printToErr();
2 F6 S! ]7 J5 w& v# b
3 R0 D- D: L' p% }

; v, C* p5 {; n
- p" B* @& e7 A4 ?; S
" L  X: o5 r1 b* S! B0 D2 q    //打印结果
" [$ Y' W+ O: K
/ G+ ^, I( b6 P    String jobName = "user defined streaming source";

) V2 S0 d4 r) Y) h5 E5 c  o3 N1 s    env.execute(jobName);
0 [0 ~- C: |' g! I( T8 V( x9 l4 ]& Y
) H5 |0 `/ H# g}
" Z' O6 o# G# i0 v. c' F) M9 o</code></pre>
<p>可以看到，我们将流进行了拆分，并且成功打印出了结果。这里要注意，Flink 最新提供的 SideOutPut 方式拆分流是<strong>可以多次进行拆分</strong>的，无需担心会爆出异常。</p>
3 b; C) q6 r! ]% {. `: M/ M+ F& F<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CBMKAGHoUAAM-5UL5geg132-20210223084827698.png" ></p>
$ b0 F) v- s9 n, {4 C<h3 id="总结">总结</h3>
<p>这一课时我们讲解了 Flink 的一个小的知识点，是我们生产实践中经常遇到的场景，Flink 在最新的版本中也推荐我们使用 SideOutPut 进行流的拆分。</p>
<blockquote>
( ?: S; F9 I2 u' w; L2 b) P<p>关注公众号：<code>大数据技术派</code>，回复<code>资料</code>，领取<code>1024G</code>资料。</p>
3 ^% G. r2 {9 z( s/ v</blockquote>
4 a+ F0 N( ~# v7 ?9 g