Flink常用API之转换算子的Split算子

Flink常用API之转换算子的Split算子

package transformationimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport source.{MyCustomerSource, StationLog}/** * @Author yqq * @Date 2021/12/26 16:51 * @Version 1.0 */object Transformation02 { def main(args: Array[String]): Unit = { val en = StreamExecutionEnvironment.getExecutionEnvironment en.setParallelism(1) import org.apache.flink.streaming.api.scala._ //读取数据源 val stream = en.addSource(new MyCustomerSource) val result: SplitStream[StationLog] = stream.split(//流并没有真被切割开 log => { if (log.callType.equals("success")) Seq("success") else Seq("no success") //给两个不同数据打上标签 } ) val stream1: DataStream[StationLog] = result.select("success") val stream2: DataStream[StationLog] = result.select("no success") stream1.print("通话成功") stream2.print("通话不成功") en.execute() }}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。