|
|
" ?: z9 T; H: X! t6 y% i. A; R<h4 id="flink系列文章">Flink系列文章</h4>
5 H p2 v. k Y2 m7 t<ol>' I. B k6 n3 z
<li><a href="https://www.ikeguang.com/?p=1976">第01讲:Flink 的应用场景和架构模型</a></li>5 Q/ X* b: T* {
<li><a href="https://www.ikeguang.com/?p=1977">第02讲:Flink 入门程序 WordCount 和 SQL 实现</a></li>
# [- n, C! r9 I: ?: Z }1 a$ M<li><a href="https://www.ikeguang.com/?p=1978">第03讲:Flink 的编程模型与其他框架比较</a></li>
7 V! J. u H" T6 g! i9 L<li><a href="https://www.ikeguang.com/?p=1982">第04讲:Flink 常用的 DataSet 和 DataStream API</a></li>
# ] a: T* n4 b$ M<li><a href="https://www.ikeguang.com/?p=1983">第05讲:Flink SQL & Table 编程和案例</a></li>
: d, g2 ]( H: K$ a9 O: V+ g<li><a href="https://www.ikeguang.com/?p=1985">第06讲:Flink 集群安装部署和 HA 配置</a></li>1 r( P+ ^) y" }$ k4 I7 Y# ~
<li><a href="https://www.ikeguang.com/?p=1986">第07讲:Flink 常见核心概念分析</a></li>
; W5 E% d0 o. C<li><a href="https://www.ikeguang.com/?p=1987">第08讲:Flink 窗口、时间和水印</a></li>
, v/ Z* o3 ~# j$ t) ~" k& H<li><a href="https://www.ikeguang.com/?p=1988">第09讲:Flink 状态与容错</a></li>) R* }2 b! w, j5 m! k
</ol>
* C( Q8 D! q2 r! Z5 K<blockquote>* p5 C+ ^$ a" P9 d2 [- z
<p>关注公众号:<code>大数据技术派</code>,回复<code>资料</code>,领取<code>1024G</code>资料。</p>
2 B7 Y# G$ l4 I) C</blockquote>/ `% ~) G& z1 R% N& ]+ O
<p>这一课时将介绍 Flink 中提供的一个很重要的功能:旁路分流器。</p>. R% M, `3 ?0 F' v9 ]
<h3 id="分流场景">分流场景</h3>
[1 S1 G- ]! ?<p>我们在生产实践中经常会遇到这样的场景,需把输入源按照需要进行拆分,比如我期望把订单流按照金额大小进行拆分,或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢?</p>3 f5 X6 u: f% B, U# y+ j
<h3 id="分流的方法">分流的方法</h3>& H. E8 ^- g* n; g" ]
<p>通常来说针对不同的场景,有以下三种办法进行流的拆分。</p>" o4 @* @! o7 a$ z
<h4 id="filter-分流">Filter 分流</h4>- A/ g8 ?9 j J3 D" J
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CAy6ADUaXAACSFUbdpuA911-20210223084827182.png" ></p>
& L5 N; C8 V' V" N: I: Q<p>Filter 方法我们在第 04 课时中(Flink 常用的 DataSet 和 DataStream API)讲过,这个算子用来根据用户输入的条件进行过滤,每个元素都会被 filter() 函数处理,如果 filter() 函数返回 true 则保留,否则丢弃。那么用在分流的场景,我们可以做多次 filter,把我们需要的不同数据生成不同的流。</p>! n7 O& J1 l9 D9 d
<p>来看下面的例子:</p>
9 K' J! j9 F5 N; O9 U2 L<p>复制代码</p>
$ }2 R& ]7 o g* U<pre><code class="language-java">public static void main(String[] args) throws Exception {9 v' K. {; g0 p) d& D2 R- F
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ \- n$ _8 ^8 N. x8 b! q+ c //获取数据源* j( H6 w- h5 j! d
List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();; f3 T* w! m' `0 W9 S1 E6 H
data.add(new Tuple3<>(0,1,0));/ v" ]% u4 o# c+ ^
data.add(new Tuple3<>(0,1,1));. V) Y6 f" U/ Z3 O s. y" D
data.add(new Tuple3<>(0,2,2));
: U9 ?; d5 X! j2 q data.add(new Tuple3<>(0,1,3));
1 z2 [( p( }# Y2 u" p% w7 t! x8 r data.add(new Tuple3<>(1,2,5));2 \4 P8 N- F+ K' a* c7 x5 B
data.add(new Tuple3<>(1,2,9));9 r: M/ d/ b% l4 X1 v- k9 O, ]5 b
data.add(new Tuple3<>(1,2,11));4 \- k/ y+ w5 A2 T T% d" O' v$ n
data.add(new Tuple3<>(1,2,13));
/ u5 k l) c3 A7 f6 Y0 E, h& J8 _/ X Z5 W/ L6 _
DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);1 D5 H* \! W5 e$ y( r
/ }( b5 m8 r9 `
4 h" M" x, o7 V( l
^# x& b, Z! v/ {5 T SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> zeroStream = items.filter((FilterFunction<Tuple3<Integer, Integer, Integer>>) value -> value.f0 == 0);& I+ y- l2 u8 B; B$ j; U* ]) }
' p Y0 h/ H7 s( c9 X ?: C
SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> oneStream = items.filter((FilterFunction<Tuple3<Integer, Integer, Integer>>) value -> value.f0 == 1);
. @/ ^ \: l \9 h2 n. X4 e/ F6 u& u! Z" k8 ?9 n1 j; L
% r6 X9 w' Z3 j I" U2 N
0 q/ u+ z% O) w z6 R
zeroStream.print();
' e9 ~: ?% I& z6 r9 l" ^
- a. u7 @$ h% \0 p( T4 b, v4 C/ d+ a oneStream.printToErr();
g7 P7 `, ^- X
% n0 m+ x; k; a* Z0 }/ F( m2 a
_6 n. D) V, b2 @; C; v8 g- O$ z& Z. z7 r
, o$ K7 V; S3 x2 k% D- s
4 u# @( _- N& j //打印结果: O3 h. @4 S+ c5 P A4 j
, k3 Z) c" j. X- Q String jobName = "user defined streaming source";
- k/ I# e, u- U/ a4 M, B, y7 ]! D4 N! l9 R; y ]% |
env.execute(jobName);
3 E$ [8 l, k' D. j9 |/ @2 o# d; b6 L- J. d$ v8 n
}2 j* F5 p3 ^; _5 k
</code></pre>" m8 k) n0 t/ w7 b# ~
<p>在上面的例子中我们使用 filter 算子将原始流进行了拆分,输入数据第一个元素为 0 的数据和第一个元素为 1 的数据分别被写入到了 zeroStream 和 oneStream 中,然后把两个流进行了打印。</p>- a; B; o9 `$ S: B
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/Ciqc1F7CA2WAYbshAAKj494h86s723-20210223084827296.png" ></p>
' }& |; x! D9 b. F: \2 B<p>可以看到 zeroStream 和 oneStream 分别被打印出来。</p>
+ z# x3 ^+ m w0 M1 _# o( B, l- K<p>Filter 的弊端是显而易见的,为了得到我们需要的流数据,需要多次遍历原始流,这样无形中浪费了我们集群的资源。</p>
& ?( W ^5 f' T<h4 id="split-分流">Split 分流</h4>) _' Z6 @0 ]( H" [* c/ I7 |* h! V
<p>Split 也是 Flink 提供给我们将流进行切分的方法,需要在 split 算子中定义 OutputSelector,然后重写其中的 select 方法,将不同类型的数据进行标记,最后对返回的 SplitStream 使用 select 方法将对应的数据选择出来。</p>1 b- s' |: a/ d' k& ?6 D
<p>我们来看下面的例子:</p>
* t6 U4 x) A: S& @+ T$ b7 v<p>复制代码</p>
5 N9 S) q0 q$ Q9 |1 A<pre><code class="language-java">public static void main(String[] args) throws Exception {
- R* s( a1 ~! M' s3 e* P2 l8 V7 h9 N" s3 P
2 c' y9 `2 N( X |6 ?
/ o% V- h$ E4 T9 \0 ?1 x, d& ~ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
' ]. l$ F1 V1 Y# y
: e4 n, @4 T% O5 b //获取数据源
! |7 C7 H) Z+ C+ w1 e0 f1 T" X; r0 q
List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();
# t/ |) C7 ~7 B' n* q' g0 Q* ~* Y
data.add(new Tuple3<>(0,1,0));+ K* Z0 _( n, p: t9 R1 x
0 M F E$ f8 t% r8 n& N, V( z. O data.add(new Tuple3<>(0,1,1));
) z" H% J3 a2 i5 x
5 ]! P. Z# i! h! N, N9 _ data.add(new Tuple3<>(0,2,2));% m6 ]" k: A" y5 N
( V1 c. [5 t6 v& r J8 V
data.add(new Tuple3<>(0,1,3));$ F9 u* C# r2 I8 j4 {
/ ^& Q6 Q. [. X* `6 x data.add(new Tuple3<>(1,2,5));6 E1 c! z& j3 q5 W& b1 t' j
1 A3 L9 i' I5 A- O; A
data.add(new Tuple3<>(1,2,9));' X- L' P( g5 e
- p3 y8 r, ?, \' q# r. c
data.add(new Tuple3<>(1,2,11));# Q, h" B, q6 `. @+ x
. S( j. `: W( i- g! L/ G
data.add(new Tuple3<>(1,2,13));3 z4 G, d' Z- w
) X) s: I, t& W2 l: u
! A* @) Y# ]9 k
% z4 B' n$ l+ s0 _' S8 P V5 A7 S0 O. {* o
, j5 B6 H. @* v4 y' y
DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);, o1 k' `( ]: q* K5 U4 Y& \+ p0 G
) k" W) ^( J7 b; B) B8 d: t0 V g8 v* L: _/ W
/ S, w0 z1 O* }% B. o3 r) D O
& z5 O4 p ~2 ~; E( f
: e* |! J5 j2 ?9 b SplitStream<Tuple3<Integer, Integer, Integer>> splitStream = items.split(new OutputSelector<Tuple3<Integer, Integer, Integer>>() {& D. E6 H7 ^$ c
C1 C; z* Z6 y @Override+ h2 N6 @, H0 N& D ~* X/ M" K
1 Q$ v' c3 ?; X8 w9 ?
public Iterable<String> select(Tuple3<Integer, Integer, Integer> value) {
# y( @# K) e4 w* N2 W% f
g7 j- q$ H0 |1 U( p; y List<String> tags = new ArrayList<>();
# j g# k+ C: S& l4 J$ D3 }3 J b5 J
if (value.f0 == 0) {0 ]( z; k! ^8 i
+ u2 v% r0 x: C, i1 c tags.add("zeroStream");4 i# ^% u/ e: a# t: K
0 D( A* @0 y2 A* B/ \" \% T } else if (value.f0 == 1) {$ l* h0 I2 b# r, y) W+ R& C4 S3 {6 f
7 u" q4 ^& M% N' E( F
tags.add("oneStream");9 t) m6 T9 X+ x7 {, z$ w6 y
- R" }2 H" w- Z9 }. n& T
} n: T/ U' W6 o1 @# E+ k
; _. L6 h; e) H% k
return tags;, A! c& L6 u, k
1 X' V; l5 M4 A* a9 w& p }7 M# p {7 _ h# `" x; z
' `8 Q `' ]- j& \0 a });! @3 h: d3 [* H# N3 Q
8 [0 a5 p1 n* p5 x. C+ l/ e# Z
- _' t% A" Z9 c2 p7 \7 W. t) @
% W% W3 T7 ^ ^+ y) G& U splitStream.select("zeroStream").print();
, o* g: m" Y1 p2 N% `- F3 r9 o, l" R, T7 S8 N7 S% h5 A" o$ m- r
splitStream.select("oneStream").printToErr();4 e3 A5 B! V* K8 {
6 u$ h6 K" ~ Q7 i+ o' \* u
' B4 G6 H1 Z9 r1 S& I* K- [) N: \+ y& `( p
//打印结果
* d: ]4 S: y! ^5 b [% l N7 Z5 K; E! |
String jobName = "user defined streaming source";
6 Z, s! w" b' O, A) o. h4 K, K+ i1 s! M
env.execute(jobName);
9 S* d# ~0 l; r9 a" f, G! c# \' U# Z8 q; ~! s; w- E, N& N
}
0 @# M% I8 @& K4 d) ?- w4 f4 k</code></pre>
) d& m! k' ?- b<p>同样,我们把来源的数据使用 split 算子进行了切分,并且打印出结果。</p>6 l+ B s+ `# [* [
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA4aAbUSJAAG1LWNB3qw627-20210223084827377.png" ></p>8 A) x8 d3 U3 t6 h/ j: k3 l* o
<p>但是要注意,使用 split 算子切分过的流,是不能进行二次切分的,假如把上述切分出来的 zeroStream 和 oneStream 流再次调用 split 切分,控制台会抛出以下异常。</p>
$ X3 L! b+ d' y# V6 |<p>复制代码</p>9 y' v4 W: t4 Q. K' N! z$ q0 G1 k
<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.
5 y+ n7 v- C( C; @5 d</code></pre>
3 O& o8 Z7 b' j/ d* y! g( g( j<p>这是什么原因呢?我们在源码中可以看到注释,该方式已经废弃并且建议使用最新的 SideOutPut 进行分流操作。</p>
. y w; q6 L# u* @# Z<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CA6OAJ-JDAAIrh1JSAEo033-20210223084827602.png" ></p>
, q/ g' N$ V8 e# r. e# a<h4 id="sideoutput-分流">SideOutPut 分流</h4>( o+ }( v& M7 M0 {# i: d4 _
<p>SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法,在使用 SideOutPut 时,需要按照以下步骤进行:</p>
% V4 \; S* M- K# A* a3 \3 F<ul>1 { y. t2 M1 D& ?' U& r( b
<li>定义 OutputTag</li>+ _7 `( C2 n- e4 `) D
<li>调用特定函数进行数据拆分
. P1 _+ I+ J! \* H2 z; N<ul>
6 f5 ]6 R: J* i; X: L. G7 i' f<li>ProcessFunction</li>
* z0 K% R9 A& c! g+ ~2 |$ O9 X3 g<li>KeyedProcessFunction</li>
! q1 V, _2 j" { d" A: p# b<li>CoProcessFunction</li>
% Y6 K% D& ^7 Y4 @" @7 Z5 B4 j<li>KeyedCoProcessFunction</li>
, D ^( q1 j" |) J9 W<li>ProcessWindowFunction</li>
1 P" }3 X) c0 |; O5 a" ^<li>ProcessAllWindowFunction</li>
( N4 X1 I" g( L! d</ul>
% F" H# ]$ V! C8 j9 s$ u</li>) o4 {+ Y7 H- n9 }, L5 H7 X0 @) V) w
</ul>$ Q' n) t* ]) Y
<p>在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut:</p>
J- h4 `' t8 K) n<p>复制代码</p># \; U8 G3 H8 D' c" s+ T
<pre><code class="language-java">public static void main(String[] args) throws Exception {
: {0 h3 J* [( c( K0 u
+ }' N3 X# s9 c
& f# v$ C8 D Z' y0 j* X6 Y4 K6 D( x7 i- d9 H( p2 H0 Q
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();) V, w& |8 Q- i: @% |) w) n
3 V. r+ [# I: S4 ~6 Y2 A& z9 ?' _ S //获取数据源
. I; w" [) r+ N- V# _" S
4 \3 L& \4 V: {7 `& z List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();4 J5 n" P9 _2 D% u# |# \! `# Y. |
i/ F, Z3 y- h- d2 x) ]' I data.add(new Tuple3<>(0,1,0));6 o6 S# k0 U' N2 {4 t9 G9 E( Q- h
, P k# ^* _2 B8 V( v6 n1 @, j
data.add(new Tuple3<>(0,1,1));
, U0 K" ]7 _, o/ ~' c/ w0 i# M1 {8 m' r9 x5 r& j4 k
data.add(new Tuple3<>(0,2,2));; P7 p% T6 }( b. p
m) m L4 t+ w N1 U data.add(new Tuple3<>(0,1,3));
# \* y9 e& {6 W2 j
+ b b# U2 Y9 j" R& M6 o, ~ data.add(new Tuple3<>(1,2,5));7 {/ R5 v5 z; Q' J/ e
# u( R3 l" g- o0 r data.add(new Tuple3<>(1,2,9));
x2 J" T) {' f* S" ~6 i1 k, g7 m/ i' a
data.add(new Tuple3<>(1,2,11));
8 C, \8 T% D/ o, l5 P. d6 d) L
, P u0 F* M3 }, R' ]1 U! h& r data.add(new Tuple3<>(1,2,13));
1 t1 V2 S( {, |) t8 [ O3 p% t
) v) f3 {6 f; a$ e _- Y( ]
: d/ A7 \" Z; P' I7 w h; A1 V% j M+ q
3 W: E# ]* D6 @8 y/ B. t' R
: W& J2 }% H. F5 q- d. e& E1 W* } DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);
, ^7 Z2 i+ c9 i3 T7 W% |& \7 A+ Y* d3 v3 s: `
5 S$ g" k) i# U* L0 I- _
- w9 j9 k9 G" E OutputTag<Tuple3<Integer,Integer,Integer>> zeroStream = new OutputTag<Tuple3<Integer,Integer,Integer>>("zeroStream") {};- H( [ d/ {3 |1 E
+ K5 Z- ^& }$ k# l6 @: o OutputTag<Tuple3<Integer,Integer,Integer>> oneStream = new OutputTag<Tuple3<Integer,Integer,Integer>>("oneStream") {};
! c% i- o5 @( q: x5 A+ M& \ V5 r" g1 O! ?
5 \9 S/ V3 G& ^# r- K8 [, ]& Q6 u! j0 I: j$ V% u6 N
$ M( K* l) p: ]5 a6 H$ A
1 y7 N) i ?8 I0 P
SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> processStream= items.process(new ProcessFunction<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {1 s' t, e0 ~# @" M
% a* `, z. u$ B# s @Override0 H4 X2 y% u; G" n
) ?: H# ^) K% h public void processElement(Tuple3<Integer, Integer, Integer> value, Context ctx, Collector<Tuple3<Integer, Integer, Integer>> out) throws Exception {
8 W+ ^. Q# }. o
! u$ B2 S% Y% U q2 M
: n+ \ g) g! J3 v3 C% F2 @3 S; z1 B% |! k# o2 J v
if (value.f0 == 0) {
2 }: s4 _2 Y+ q( U# c3 ~5 K3 J) h! G1 g2 P5 ^& X
ctx.output(zeroStream, value);
: W# ?2 t8 r t$ g' `7 I4 f
5 p7 U4 m2 ?9 a9 a" L; z" Y } else if (value.f0 == 1) {5 p6 Y' e8 Y9 B: D4 O
% h2 y/ q- J; m ctx.output(oneStream, value);
b/ k; E8 _6 s8 O
: L- ] B$ p% l% n7 @4 H }) C4 L4 F: F; N6 j4 `$ b
2 W1 P, h3 z5 L* E2 `4 S* [7 I# s }! @; ^ Q3 o* N, |# F
% x' o7 \8 |6 X
});: `; a3 N, x8 h5 i# h. d$ O
5 @; h6 T7 U6 r$ }% {) h! x6 }
& H A& M1 q9 i. M
/ v, _, u) q3 |. `5 `
DataStream<Tuple3<Integer, Integer, Integer>> zeroSideOutput = processStream.getSideOutput(zeroStream);2 D1 [/ g. t. Q3 e) H5 p
5 Y4 y4 ^+ V4 g0 y* D: z DataStream<Tuple3<Integer, Integer, Integer>> oneSideOutput = processStream.getSideOutput(oneStream);
2 v* l. d- ^5 G2 F
: b% M9 d1 O2 ]: p/ |! {
# K- f% N! @! @7 P: Z; k9 G; d* c) T& m; {+ W
zeroSideOutput.print(); W9 z- F* c2 G/ J- o5 [8 r+ Q4 w
% @, i0 I5 t4 z8 ?
oneSideOutput.printToErr();6 ?! M; i# r& [# J: {+ ^ s% z. u
9 l K& n O# v/ p$ ~$ Y
! s$ o: M4 m0 {9 ?2 I, O
9 Q- g. G3 _! C3 |. F
6 B9 P! Y: | G& w
/ |+ A. u) P2 b% `9 W2 |( x //打印结果
8 m& |7 O5 I5 b2 i2 C0 R) h
$ ~1 S. S0 Z7 H4 a) D String jobName = "user defined streaming source";7 ^) r N' [7 Q' x0 t+ e$ _
* _$ S* x1 u9 l5 @
env.execute(jobName);% X1 Y3 @% y9 I
* e2 h# L- s5 h9 d' j. h3 v4 Q
}$ f9 V" }) L4 O
</code></pre>
: |2 Q, @; n4 x, r3 o5 k' C<p>可以看到,我们将流进行了拆分,并且成功打印出了结果。这里要注意,Flink 最新提供的 SideOutPut 方式拆分流是<strong>可以多次进行拆分</strong>的,无需担心会爆出异常。</p>( P. K# V' | P7 O5 c
<p><img src="https://kingcall.oss-cn-hangzhou.aliyuncs.com/blog/img/CgqCHl7CBMKAGHoUAAM-5UL5geg132-20210223084827698.png" ></p>
^: S3 T; C W& w0 E9 Q0 T<h3 id="总结">总结</h3>
2 h3 t. D# a' h& a# o4 m; K<p>这一课时我们讲解了 Flink 的一个小的知识点,是我们生产实践中经常遇到的场景,Flink 在最新的版本中也推荐我们使用 SideOutPut 进行流的拆分。</p>, Q& a7 O9 R. `! [: d7 w% q$ z! c4 ]$ d
<blockquote>
, f5 ?. n+ P. l. X<p>关注公众号:<code>大数据技术派</code>,回复<code>资料</code>,领取<code>1024G</code>资料。</p>
* T! N" V7 X/ K, P, u8 g: T$ r</blockquote>, Y3 Q8 S g& @7 P
0 d0 f4 n) [9 _" F/ d |
|